From 839bc4a6e5cbfe3866badcc090f559d3a7cec645 Mon Sep 17 00:00:00 2001 From: jochen Date: Wed, 11 Nov 2026 11:44:42 +0100 Subject: [PATCH 01/10] TL0: the teller's test follows the contracts, the settle ask gone with the verb settle MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Builds: design 47 G §431 (the settle ask goes with the verb settle) Claude-Session: https://claude.ai/code/session_01JG73KEq4zoDfM9WtYpcQgZ --- .../cmd/mesh-delivery-teller/teller.go | 4 ++-- .../cmd/mesh-delivery-teller/teller_test.go | 9 ++++----- 2 files changed, 6 insertions(+), 7 deletions(-) diff --git a/modules/mesh-delivery-teller/cmd/mesh-delivery-teller/teller.go b/modules/mesh-delivery-teller/cmd/mesh-delivery-teller/teller.go index 83506c23..4426cb51 100644 --- a/modules/mesh-delivery-teller/cmd/mesh-delivery-teller/teller.go +++ b/modules/mesh-delivery-teller/cmd/mesh-delivery-teller/teller.go @@ -324,8 +324,8 @@ func (t *Teller) catchUp(id string) { t.said(h, *w, err) } -// mergeAct is what the approve option of the gate's ask binds: the ask's verb (meshdeliveryclient.AskVerb: mesh-delivery.merge, or -// mesh-delivery.settle for a person's settle ask, novox/hq ADR 0556 clause 5), this ask, the answer, and the heads it covers. +// mergeAct is what the approve option of the gate's ask binds: the ask's verb (meshdeliveryclient.AskVerb: mesh-delivery.merge; the settle ask +// went with the verb settle, design 47 G §431), this ask, the answer, and the heads it covers. // It is spelled as the gate composes it (mesh-catalog modules/mesh-delivery-gate approval.go mergeAct, from // modules/mesh-delivery approval.go:63-65) and as mesh-controller verifies it (cmd/mesh-controller/moved.go:98-106). func mergeAct(verb, ask, option string, heads map[string]string) occ.Act { diff --git a/modules/mesh-delivery-teller/cmd/mesh-delivery-teller/teller_test.go b/modules/mesh-delivery-teller/cmd/mesh-delivery-teller/teller_test.go index 34758645..dac5305b 100644 --- a/modules/mesh-delivery-teller/cmd/mesh-delivery-teller/teller_test.go +++ b/modules/mesh-delivery-teller/cmd/mesh-delivery-teller/teller_test.go @@ -412,16 +412,15 @@ func TestAWarrantThatHoldsWritesOneAnswerAndASecondWritesNone(t *testing.T) { // if len(b.conds) != 0 { t.Errorf("conditions said: %+v", b.conds) } - // 1280 (ADR 0556 clause 5): a settle ask holds when its options bind mesh-delivery.settle, and not when they bind a merge. - for verb, holds := range map[string]bool{"mesh-delivery.settle": true, "mesh-delivery.merge": false} { + // design 47 G §431: the settle ask went with the verb settle, so an ask whose options bind it holds no more. + for verb, holds := range map[string]bool{"mesh-delivery.settle": false, "mesh-delivery.merge": true} { b := newBed(t) - group := meshdeliveryclient.SettleGroup([]string{"novox/mesh-delivery@" + headA[:12]}) - key, value, ga := gateAskOf(t, group, verb, twoHeads(), twoHeads(), 1) + key, value, ga := gateAskOf(t, "g", verb, twoHeads(), twoHeads(), 1) b.gate[key] = value b.t.OnAsk(key, 1, value) b.deliver(t, warrantFor(ga.Ask, "approve")) if _, written := b.answers.written[answerKey(key)]; written != holds { - t.Errorf("a settle ask binding %s: answer written %v, want %v", verb, written, holds) + t.Errorf("an ask binding %s: answer written %v, want %v", verb, written, holds) } } } From 154505f451ae9da7f6a8ad2e2ee0f58031cca04d Mon Sep 17 00:00:00 2001 From: jochen Date: Wed, 11 Nov 2026 11:45:37 +0100 Subject: [PATCH 02/10] Assigner: repair against the new contracts (source key per branch, want with digests, seat verbs not yet served) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Builds: design 47 G §15, §95, §353, §379 Claude-Session: https://claude.ai/code/session_01JG73KEq4zoDfM9WtYpcQgZ --- .../cmd/mesh-delivery-assigner/pass_test.go | 8 ++++---- .../cmd/mesh-delivery-assigner/wants_tools.go | 3 +++ .../cmd/mesh-delivery-assigner/world.go | 5 ++++- .../mesh-delivery-assigner/internal/assign/wants_test.go | 3 ++- 4 files changed, 13 insertions(+), 6 deletions(-) diff --git a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/pass_test.go b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/pass_test.go index 5b077f78..0bc60e9a 100644 --- a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/pass_test.go +++ b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/pass_test.go @@ -1128,10 +1128,10 @@ func aReleasesApproveAnswersTheMoveOfItsBuildAlone(t *testing.T) { p.logf = t.Logf mod, _ := meshcatalogclient.EncodeModule(meshcatalog.ModulesByModuleValue{Fp2: fpTest, Repository: "novox/web", Commit: commit, Src: 7, FpSrc: c.fpSrc, Path: meshcatalogclient.RootPath}) p.w.put(buckets.Modules, "web", mod) - src, _ := mdc.EncodeSource(mdseat.SourceByMainValue{Group: "feat/web", Heads: map[string]string{"novox/web": commit}}) - p.w.putEntry(buckets.Source, mdc.SourceMainKey, src, 7, time.Now()) - later, _ := mdc.EncodeSource(mdseat.SourceByMainValue{Group: "feat/other", Heads: map[string]string{"novox/web": commit, "novox/x": commit}}) - p.w.putEntry(buckets.Source, mdc.SourceMainKey, later, 8, time.Now()) // a later merge leaves revision 7's release + src, _ := mdc.EncodeSource(mdseat.SourceByBranchValue{Group: "feat/web", Heads: map[string]string{"novox/web": commit}}) + p.w.putEntry(buckets.Source, mdc.SourceKey(trunkBranch), src, 7, time.Now()) + later, _ := mdc.EncodeSource(mdseat.SourceByBranchValue{Group: "feat/other", Heads: map[string]string{"novox/web": commit, "novox/x": commit}}) + p.w.putEntry(buckets.Source, mdc.SourceKey(trunkBranch), later, 8, time.Now()) // a later merge leaves revision 7's release if c.mergedAt != "" { merge, err := mdc.EncodeMerge(mdseat.MergesByGroupValue{Group: "feat/web", Ask: "merge.feat_web.0123456789ab", Pulls: []mdseat.MergesByGroupValuePullsItem{ {Repository: "novox/web", Number: 3, Head: strings.Repeat("3", 40), State: string(mdc.MergeMerged), Commit: c.mergedAt}}}) diff --git a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/wants_tools.go b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/wants_tools.go index 02a0d387..034e9b15 100644 --- a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/wants_tools.go +++ b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/wants_tools.go @@ -315,6 +315,9 @@ func (s *wantsTools) tools() []stdio.Tool { } var out []stdio.Tool for _, v := range meshwants.Verbs() { + if run[v] == nil { + continue // a verb of the seat this module does not serve yet: revision, entry, difference (AS11), input (AS5) + } out = append(out, stdio.Tool{Name: meshwants.Seat + "." + v, Run: run[v]}) } return out diff --git a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/world.go b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/world.go index cf2b67f4..893d02f5 100644 --- a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/world.go +++ b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/world.go @@ -86,7 +86,7 @@ func (w *world) putEntry(bucket, key string, value []byte, revision uint64, at t if bucket == plc.AssignmentsBucket { _, _ = w.assignmentKeys.Put(key, value, revision, false) // put took it already; this is its revision } - if bucket == buckets.Source && key == mdc.SourceMainKey { + if bucket == buckets.Source && key == mdc.SourceKey(trunkBranch) { if s, err := mdc.DecodeSource(value); err == nil && s.Group != "" { w.sourceGroups[revision] = s.Group } @@ -575,3 +575,6 @@ func (w *world) unplacedNodes() []string { sort.Strings(out) return out } + +// trunkBranch is the branch of the trunk-only want, kept until the first registrations (design 47 G §95, §353). +const trunkBranch = "main" diff --git a/modules/mesh-delivery-assigner/internal/assign/wants_test.go b/modules/mesh-delivery-assigner/internal/assign/wants_test.go index d15823b1..1c38e3a7 100644 --- a/modules/mesh-delivery-assigner/internal/assign/wants_test.go +++ b/modules/mesh-delivery-assigner/internal/assign/wants_test.go @@ -9,6 +9,7 @@ import ( "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-builder/builds" "git.novox.be/novox/mesh-delivery/modules/mesh-grants/grantsstate" "mesh-delivery-assigner/internal/puts" + "reflect" "slices" "strings" "testing" @@ -431,7 +432,7 @@ func TestEachRuleHoldsTheWantBack(t *testing.T) { before["g14"] = mwc.Want{Fp: fNew, Composed: dig(fNew)} } out := Wants(m, n, before, nil) - if got := out.Wants["g14"]; got != c.want { + if got := out.Wants["g14"]; !reflect.DeepEqual(got, c.want) { t.Errorf("%s: g14's want = %+v, want %+v", c.name, got, c.want) } if got := out.Rules["g14"].Rule; got != c.rule { From 18c954ddcd2d0d8167101a0e88e656a875675752 Mon Sep 17 00:00:00 2001 From: jochen Date: Wed, 11 Nov 2026 11:45:50 +0100 Subject: [PATCH 03/10] CT10: meshwantsclient holds a part's view of the assignment half, a first view then assignment-revision MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The typed first view and the following of each half sit in the seats' client packages, built on go/views (design 47 G §245), so no part follows the assignment half by a follower of its own: a newer trunk revision heard is waited for, said in the status, asked and held; a candidate's, or one held, asks nothing. Builds: design 47 G §245, §365 Claude-Session: https://claude.ai/code/session_01JG73KEq4zoDfM9WtYpcQgZ --- .../go/clients/meshwantsclient/assignment.go | 176 ++++++++++++++++++ .../meshwantsclient/assignment_test.go | 116 ++++++++++++ contracts/go/clients/meshwantsclient/go.mod | 9 + contracts/go/clients/meshwantsclient/go.sum | 24 +++ 4 files changed, 325 insertions(+) create mode 100644 contracts/go/clients/meshwantsclient/assignment.go create mode 100644 contracts/go/clients/meshwantsclient/assignment_test.go create mode 100644 contracts/go/clients/meshwantsclient/go.sum diff --git a/contracts/go/clients/meshwantsclient/assignment.go b/contracts/go/clients/meshwantsclient/assignment.go new file mode 100644 index 00000000..19a716b1 --- /dev/null +++ b/contracts/go/clients/meshwantsclient/assignment.go @@ -0,0 +1,176 @@ +package meshwantsclient + +import ( + "context" + "encoding/json" + "fmt" + "strconv" + "sync" + "time" + + seat "git.novox.be/novox/mesh-delivery/contracts/go/seats/meshwants" + "git.novox.be/novox/mesh-sdk/go/buscalls" + "git.novox.be/novox/mesh-sdk/go/views" +) + +// RevisionVerb is the seat verb a part takes its first view of the assignment half by, as its manifest's invokes names +// it. +const RevisionVerb = "seat:" + seat.Seat + "." + seat.VerbRevision + +// kindTrunk is the kind of an assignment graph revision a running part acts on; a candidate's moves nothing. +const kindTrunk = "trunk" + +// 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) + +// assignmentCreated 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 assignmentCreated = time.Unix(0, 0).UTC() + +// assignmentViewName is how an assignment half's revision is named in a part's first-view status (views.Status.View, +// WaitsFor). +func assignmentViewName(revision int64) string { return strconv.FormatInt(revision, 10) } + +// Assignment is a part's view of the assignment half (design 47 G §245): a first view of the newest trunk revision asked +// of the seat mesh-wants' verb `revision`, then kept from its event `assignment-revision`, 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 `.`, each value that entry's reference. With no view the part acts on nothing: +// Revision says false, and a revision heard and not held yet is waited for, said in the status, and past +// views.WaitsBound raised as view-waits. +type Assignment struct { + Follower *views.Follower + // Held, when set, is told of each revision once the view holds it, on the ask's goroutine and outside any lock. + Held func(seat.AssignmentRevision) + + call Call + mu sync.Mutex + held *seat.AssignmentRevision + waits int64 +} + +// NewAssignment is an Assignment asking the seat through call. Nothing is asked before Start. +func NewAssignment(call Call) *Assignment { + a := &Assignment{call: call} + a.Follower = &views.Follower{Owner: seat.Seat, Tool: RevisionVerb, State: seat.StateAssignmentRevisions, Ask: a.ask} + return a +} + +// WireAssignment is an Assignment adopted by a stdio part's views.Reader: the reader writes its conditions and its +// view-waits finding, reads it as `mesh-wants.assignment-revisions`, lists and asks it again in its first-view tools, and +// at its Start subscribes the seat's event assignment-revision (the part's manifest consumes it) into Hear and asks the +// first view. held, when not nil, is the Assignment's Held. The part declares the seat's verb revision in invokes. +func WireAssignment(r *views.Reader, call Call, held func(seat.AssignmentRevision)) (*Assignment, error) { + a := NewAssignment(call) + a.Held = held + if err := r.Adopt(a.Follower, seat.Seat, seat.EventAssignmentRevision, a.Hear); err != nil { + return nil, err + } + return a, nil +} + +// WireAssignmentProcess is WireAssignment for a process's views.ProcessReader: its picture +// (`mesh-wants_assignment-revisions`) reads the half as a read-only bucket, and the seat's event assignment-revision on +// the module's durable consumer goes to Hear. +func WireAssignmentProcess(r *views.ProcessReader, call Call, held func(seat.AssignmentRevision)) (*Assignment, *views.Picture, error) { + a := NewAssignment(call) + a.Held = held + p, err := r.Adopt(a.Follower, seat.Seat, seat.EventAssignmentRevision, a.Hear) + if err != nil { + return nil, nil, err + } + return a, p, nil +} + +// Start asks the first view, the newest trunk revision, until it is answered or ctx ends. +func (a *Assignment) Start(ctx context.Context) error { return a.Follower.Start(ctx) } + +// Revision is the assignment graph revision the view holds; false while it has none, when the part acts on nothing new. +func (a *Assignment) Revision() (seat.AssignmentRevision, bool) { + a.mu.Lock() + defer a.mu.Unlock() + if a.held == nil { + return seat.AssignmentRevision{}, false + } + return *a.held, true +} + +// Hear takes one `assignment-revision` event, as the part's durable consumer of the seat's events hands it. A +// candidate's revision moves nothing; a trunk revision the view holds, or one older, is passed over; a newer one is +// waited for and asked 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 (a *Assignment) Hear(ctx context.Context, data []byte) error { + var ev seat.AssignmentRevisionEvent + if err := json.Unmarshal(data, &ev); err != nil || ev.Revision <= 0 { + return fmt.Errorf("meshwantsclient: not a %s event: %s", seat.EventAssignmentRevision, clipBytes(data)) + } + if ev.Kind != kindTrunk { + return nil + } + a.mu.Lock() + held := int64(0) + if a.held != nil { + held = a.held.Revision + } + if ev.Revision > a.waits { + a.waits = ev.Revision + } + a.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. + a.Follower.Apply(views.Event{Owner: seat.Seat, State: seat.StateAssignmentRevisions, Revision: uint64(ev.Revision), Created: assignmentCreated, Gap: true}) + a.Follower.WaitFor(ctx, assignmentViewName(ev.Revision)) + go func() { + _ = a.Follower.Gap(ctx, fmt.Sprintf("assignment revision %d said, the view holds %d", ev.Revision, held)) + }() + return nil +} + +// ask is the Follower's Ask: the newest trunk revision, whole in one answer (its entries are named by digest). +func (a *Assignment) ask(_ context.Context, _ string) (views.Answer, error) { + raw, err := a.call(RevisionVerb, seat.RevisionInput{}) + if err != nil { + return views.Answer{}, err + } + answer, said, ok := buscalls.Unwrap(raw) + if !ok { + return views.Answer{}, fmt.Errorf("%s answered: %s", RevisionVerb, said) + } + var ans seat.RevisionAnswer + if err := json.Unmarshal(answer, &ans); err != nil { + return views.Answer{}, fmt.Errorf("%s answered what is not its answer: %s", RevisionVerb, clipBytes(answer)) + } + r := ans.Revision + if r.Revision <= 0 || r.Kind != kindTrunk { + return views.Answer{}, fmt.Errorf("%s answered a %q revision %d for the trunk", RevisionVerb, r.Kind, r.Revision) + } + values := make(map[string]json.RawMessage, len(r.Entries)) + for _, e := range r.Entries { + v, _ := json.Marshal(e) + values[e.Node+"."+e.Module] = v + } + a.mu.Lock() + moved := a.held == nil || a.held.Revision != r.Revision + a.held = &r + holds := a.waits <= r.Revision + if holds { + a.waits = 0 + } + a.mu.Unlock() + if holds { + a.Follower.Holds(assignmentViewName(r.Revision)) + } + if moved && a.Held != nil { + a.Held(r) + } + return views.Answer{V: 1, Owner: seat.Seat, State: seat.StateAssignmentRevisions, Revision: uint64(r.Revision), Created: assignmentCreated, 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/meshwantsclient/assignment_test.go b/contracts/go/clients/meshwantsclient/assignment_test.go new file mode 100644 index 00000000..9d511a02 --- /dev/null +++ b/contracts/go/clients/meshwantsclient/assignment_test.go @@ -0,0 +1,116 @@ +package meshwantsclient + +import ( + "context" + "encoding/json" + "fmt" + "sync" + "testing" + "time" + + seat "git.novox.be/novox/mesh-delivery/contracts/go/seats/meshwants" +) + +// fakeWants answers the verb revision with its newest trunk revision, and counts each ask. +type fakeWants struct { + mu sync.Mutex + newest int64 + asks int +} + +func (f *fakeWants) call(key string, body any) (json.RawMessage, error) { + if key != RevisionVerb { + return nil, fmt.Errorf("asked %s", key) + } + if in := body.(seat.RevisionInput); in != (seat.RevisionInput{}) { + return nil, fmt.Errorf("asked %+v", in) + } + f.mu.Lock() + defer f.mu.Unlock() + f.asks++ + r := seat.AssignmentRevision{Revision: f.newest, Kind: kindTrunk, Placements: "p", + Catalogs: []seat.CatalogOf{{Branch: "main", Revision: 10 + f.newest}}, + Entries: []seat.NodeEntryRef{{Node: "anchor", Module: "nats", Digest: fmt.Sprintf("d%d", f.newest)}}} + return json.Marshal(seat.RevisionAnswer{Revision: r}) +} + +func (f *fakeWants) set(newest int64) { + f.mu.Lock() + defer f.mu.Unlock() + f.newest = newest +} + +func (f *fakeWants) asked() int { + f.mu.Lock() + defer f.mu.Unlock() + return f.asks +} + +func revisionEvent(rev int64, kind string) []byte { + b, _ := json.Marshal(seat.AssignmentRevisionEvent{V: 1, Revision: rev, Kind: kind}) + return b +} + +// CT10 (design 47 G §245): a part takes a first view of the assignment half by the verb revision, then follows its +// event assignment-revision: a newer trunk revision is waited for, said in the status, and held; a candidate's, or one +// held, asks nothing; what is no such event is refused. +func TestTheAssignmentHalfIsAFirstViewThenItsEvents(t *testing.T) { + ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second) + defer cancel() + held := make(chan seat.AssignmentRevision, 8) + next := func() seat.AssignmentRevision { + t.Helper() + select { + case r := <-held: + return r + case <-ctx.Done(): + t.Fatal("no revision held") + return seat.AssignmentRevision{} + } + } + f := &fakeWants{newest: 3} + a := NewAssignment(f.call) + a.Follower.Wait = time.Nanosecond + a.Held = func(r seat.AssignmentRevision) { held <- r } + if _, ok := a.Revision(); ok { + t.Fatal("a revision before its first view") + } + if err := a.Start(ctx); err != nil { + t.Fatal(err) + } + if r := next(); r.Revision != 3 || len(r.Catalogs) != 1 || r.Catalogs[0].Revision != 13 { + t.Fatalf("first view holds %+v", r) + } + if v, ok := a.Follower.Get("anchor.nats"); !ok || string(v) == "" { + t.Errorf("the picture lacks anchor.nats: %s", v) + } + + for _, ev := range [][]byte{revisionEvent(3, kindTrunk), revisionEvent(2, kindTrunk), revisionEvent(9, "candidate")} { + if err := a.Hear(ctx, ev); err != nil { + t.Fatal(err) + } + } + if n := f.asked(); n != 1 { + t.Errorf("a revision held, an older one or a candidate's asked again: %d asks", n) + } + + f.set(4) + if err := a.Hear(ctx, revisionEvent(4, kindTrunk)); err != nil { + t.Fatal(err) + } + if r := next(); r.Revision != 4 { + t.Fatalf("after assignment-revision 4 the view holds %d", r.Revision) + } + if s := a.Follower.Status(); s.View != "4" || s.WaitsFor != "" { + t.Errorf("status view %q, waits for %q", s.View, s.WaitsFor) + } + if r, ok := a.Revision(); !ok || r.Revision != 4 { + t.Errorf("Revision = %d, %v", r.Revision, ok) + } + + for _, data := range []string{`{}`, `not json`, `{"kind":"trunk"}`} { + if err := a.Hear(ctx, []byte(data)); err == nil { + t.Errorf("Hear(%s) took it", data) + } + } +} diff --git a/contracts/go/clients/meshwantsclient/go.mod b/contracts/go/clients/meshwantsclient/go.mod index 470f8994..88f343dc 100644 --- a/contracts/go/clients/meshwantsclient/go.mod +++ b/contracts/go/clients/meshwantsclient/go.mod @@ -6,3 +6,12 @@ require ( git.novox.be/novox/mesh-delivery/contracts/go/seats/meshwants 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/meshwantsclient/go.sum b/contracts/go/clients/meshwantsclient/go.sum new file mode 100644 index 00000000..4ac8a82c --- /dev/null +++ b/contracts/go/clients/meshwantsclient/go.sum @@ -0,0 +1,24 @@ +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/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/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/nkeys v0.4.16 h1:rd5oAuLOb8mnAycB0xleuEBNS1pVVnN0fv/FF34Eypg= +github.com/nats-io/nuid v1.0.1/go.mod h1:19wcPz3Ph3q0Jbyiqsd0kePYG7A95tJPxeL+1OSON2c= +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/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= From 9b9e9f68bb538c4938adb036c50981d5a696d927 Mon Sep 17 00:00:00 2001 From: jochen Date: Wed, 11 Nov 2026 11:48:34 +0100 Subject: [PATCH 04/10] TL4: the teller's manifest says version 0.1, declares its finding stream and hears the keeper's ask MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Builds: design 47 G §383, §387, §368, §411 Claude-Session: https://claude.ai/code/session_01JG73KEq4zoDfM9WtYpcQgZ --- modules/mesh-delivery-teller/manifest_test.go | 22 ++++++++++++++++--- modules/mesh-delivery-teller/module.json | 11 +++++++++- 2 files changed, 29 insertions(+), 4 deletions(-) diff --git a/modules/mesh-delivery-teller/manifest_test.go b/modules/mesh-delivery-teller/manifest_test.go index 829136df..953960be 100644 --- a/modules/mesh-delivery-teller/manifest_test.go +++ b/modules/mesh-delivery-teller/manifest_test.go @@ -23,6 +23,13 @@ type state struct { History int `json:"history,omitempty"` } +type stream struct { + Name string `json:"name"` + Events []string `json:"events"` + MaxAgeDays int `json:"max-age-days"` + MaxMiB int `json:"max-mib"` +} + type artifact struct { Name string `json:"name"` Kind string `json:"kind"` @@ -44,6 +51,7 @@ type manifest struct { Invokes []string `json:"invokes"` Consumes []string `json:"consumes"` Emits []string `json:"emits"` + Streams []stream `json:"streams"` OwnSecrets map[string]string `json:"own-secrets"` SecretsOwner string `json:"secrets-owner"` Resources []map[string]any `json:"resources"` @@ -102,7 +110,8 @@ func TestManifestDeclarations(t *testing.T) { // 1240/AC1 if want := []string{"mesh-delivery-assigner.assigner_read_asks", "seat:mesh-hand-acts.hand-act"}; !reflect.DeepEqual(m.Invokes, want) { t.Errorf("invokes = %v, want %v", m.Invokes, want) } - if want := []string{"mesh-delivery-assigner.views.asks"}; !reflect.DeepEqual(m.Consumes, want) { + // design 47 G §387: the keeper's ask for the findings is heard, and the finding stream is declared (§383). + if want := []string{"mesh-conditions.findings-asked", "mesh-delivery-assigner.views.asks"}; !reflect.DeepEqual(m.Consumes, want) { t.Errorf("consumes = %v, want %v", m.Consumes, want) } if want := []string{"operator-channel"}; !reflect.DeepEqual(sorted(m.Uses), want) { @@ -122,8 +131,15 @@ func TestManifestDeclarations(t *testing.T) { // 1240/AC1 if want := []string{tellerevents.AskEnded}; !reflect.DeepEqual(m.Emits, want) { t.Errorf("emits = %v, want %v", m.Emits, want) } - // No streams, bus or seat (ADR 0437 §4). - for _, k := range []string{"streams", "logs", "claims", "provides", "requires", "defines-seats", "seat", "seats", "bus"} { + // design 47 G §383, §387: the stream -findings, 7 days and 16 MiB; §368: version is MAJOR.MINOR. + if want := []stream{{Name: "mesh-delivery-teller-findings", Events: []string{"finding-written.*", "finding-cleared.*", "findings-stated.*"}, MaxAgeDays: 7, MaxMiB: 16}}; !reflect.DeepEqual(m.Streams, want) { + t.Errorf("streams = %+v, want %+v", m.Streams, want) + } + if m.Version != "0.1" { + t.Errorf("version = %q, want 0.1", m.Version) + } + // No bus or seat (ADR 0437 §4). + for _, k := range []string{"logs", "claims", "provides", "requires", "defines-seats", "seat", "seats", "bus"} { if _, ok := top[k]; ok { t.Errorf("the manifest declares %q: the teller has none", k) } diff --git a/modules/mesh-delivery-teller/module.json b/modules/mesh-delivery-teller/module.json index c6e1a80c..cc176fd7 100644 --- a/modules/mesh-delivery-teller/module.json +++ b/modules/mesh-delivery-teller/module.json @@ -1,6 +1,6 @@ { "module": "mesh-delivery-teller", - "version": "1", + "version": "0.1", "slug": "teller", "tools": [ "teller_end_ask", @@ -12,8 +12,17 @@ "seat:mesh-hand-acts.hand-act" ], "consumes": [ + "mesh-conditions.findings-asked", "mesh-delivery-assigner.views.asks" ], + "streams": [ + { + "name": "mesh-delivery-teller-findings", + "events": ["finding-written.*", "finding-cleared.*", "findings-stated.*"], + "max-age-days": 7, + "max-mib": 16 + } + ], "emits": [ "ask-ended" ], From 799241364fea7d581c67cf9a7988c6f6192c5653 Mon Sep 17 00:00:00 2001 From: jochen Date: Wed, 11 Nov 2026 11:50:04 +0100 Subject: [PATCH 05/10] VW0: the view reads the trunk's source by the branch key and the per-branch value MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Builds: design 47 G §305 (the state source is kept per watched branch), §431 Claude-Session: https://claude.ai/code/session_01JG73KEq4zoDfM9WtYpcQgZ --- .../cmd/mesh-delivery-view/integration_test.go | 4 ++-- modules/mesh-delivery-view/internal/view/names.go | 5 +++++ modules/mesh-delivery-view/internal/view/reach.go | 2 +- modules/mesh-delivery-view/internal/view/reach_test.go | 6 +++--- modules/mesh-delivery-view/internal/view/tiers.go | 3 +-- modules/mesh-delivery-view/internal/view/view.go | 2 +- modules/mesh-delivery-view/internal/view/wants_test.go | 4 ++-- 7 files changed, 15 insertions(+), 11 deletions(-) diff --git a/modules/mesh-delivery-view/cmd/mesh-delivery-view/integration_test.go b/modules/mesh-delivery-view/cmd/mesh-delivery-view/integration_test.go index cd326086..77ad3d76 100644 --- a/modules/mesh-delivery-view/cmd/mesh-delivery-view/integration_test.go +++ b/modules/mesh-delivery-view/cmd/mesh-delivery-view/integration_test.go @@ -797,7 +797,7 @@ func TestProcessWholeSaysTheSourceRevisionIsDelivered(t *testing.T) { want, _ := meshwantsclient.EncodeWant(meshwantsclient.Want{Fp: fp, Composed: "sha256:" + strings.Repeat("c", 64)}) failed, _ := nodeengineclient.EncodeHave(nodeengine.HaveByNodeModuleValue{Fp: fp, Phase: nodeengineclient.PhaseRunning, Health: nodeengineclient.HealthFailed}) good, _ := nodeengineclient.EncodeHave(nodeengine.HaveByNodeModuleValue{Fp: fp, Phase: nodeengineclient.PhaseRunning, Health: nodeengineclient.HealthOK}) - src, _ := meshdeliveryclient.EncodeSource(meshdelivery.SourceByMainValue{Group: "feat/x", Heads: map[string]string{"novox/web": strings.Repeat("1", 40)}}) + src, _ := meshdeliveryclient.EncodeSource(meshdelivery.SourceByBranchValue{Group: "feat/x", Heads: map[string]string{"novox/web": strings.Repeat("1", 40)}}) for bucket, kv := range map[string][]byte{view.ModulesBucket: mod} { if _, err := kvs[bucket].Put(ctx, "web", kv); err != nil { t.Fatal(err) @@ -809,7 +809,7 @@ func TestProcessWholeSaysTheSourceRevisionIsDelivered(t *testing.T) { if _, err := kvs[view.HaveBucket].Put(ctx, "ace.web", failed); err != nil { t.Fatal(err) } - rev, err := kvs[view.SourceBucket].Put(ctx, meshdeliveryclient.SourceMainKey, src) + rev, err := kvs[view.SourceBucket].Put(ctx, meshdeliveryclient.SourceKey("main"), src) if err != nil { t.Fatal(err) } diff --git a/modules/mesh-delivery-view/internal/view/names.go b/modules/mesh-delivery-view/internal/view/names.go index 0fc71ca2..4a2d62f1 100644 --- a/modules/mesh-delivery-view/internal/view/names.go +++ b/modules/mesh-delivery-view/internal/view/names.go @@ -3,6 +3,7 @@ package view import ( "fmt" + "git.novox.be/novox/mesh-delivery/contracts/go/clients/meshdeliveryclient" "git.novox.be/novox/mesh-delivery/contracts/go/seats/meshcatalog" "git.novox.be/novox/mesh-delivery/sdk/trunk" "git.novox.be/novox/mesh-sdk/go/keys" @@ -88,3 +89,7 @@ func PackagesKey(module string) (string, error) { } return meshcatalog.PackagesByModuleKey(keys.EscapeName(module)), nil } + +// trunkSourceKey is the key of the gate's state source for the trunk, the watched branch `main` (design 47 G §305: the source +// is kept per watched branch, meshdeliveryclient.SourceKey; the view's reads are of the trunk's). +var trunkSourceKey = meshdeliveryclient.SourceKey("main") diff --git a/modules/mesh-delivery-view/internal/view/reach.go b/modules/mesh-delivery-view/internal/view/reach.go index b11e0648..d74e058c 100644 --- a/modules/mesh-delivery-view/internal/view/reach.go +++ b/modules/mesh-delivery-view/internal/view/reach.go @@ -71,7 +71,7 @@ func (m *Model) applyHistory(e Entry) { } m.modHist[e.Key] = h case SourceBucket: - if e.Key != meshdeliveryclient.SourceMainKey || e.Deleted || e.Revision == 0 { + if e.Key != trunkSourceKey || e.Deleted || e.Revision == 0 { return } src, err := meshdeliveryclient.DecodeSource(e.Value) diff --git a/modules/mesh-delivery-view/internal/view/reach_test.go b/modules/mesh-delivery-view/internal/view/reach_test.go index 12e32744..5d53d583 100644 --- a/modules/mesh-delivery-view/internal/view/reach_test.go +++ b/modules/mesh-delivery-view/internal/view/reach_test.go @@ -49,11 +49,11 @@ func TestTheViewDerivesHowFarEachRepositoryReached(t *testing.T) { return append([]Entry{modIn(fpLater, commit, 12, to, true), modIn(fpLater, commit, 12, to, false)}, extra...) } src := func(rev uint64, webHead string) Entry { - v, err := meshdeliveryclient.EncodeSource(meshdelivery.SourceByMainValue{Group: "g", Heads: map[string]string{web: webHead, sdk: commitS}}) + v, err := meshdeliveryclient.EncodeSource(meshdelivery.SourceByBranchValue{Group: "g", Heads: map[string]string{web: webHead, sdk: commitS}}) if err != nil { t.Fatal(err) } - return Entry{Bucket: SourceBucket, Key: meshdeliveryclient.SourceMainKey, Value: v, At: t0, Revision: rev, History: true} + return Entry{Bucket: SourceBucket, Key: trunkSourceKey, Value: v, At: t0, Revision: rev, History: true} } taken := func(repository, merge string) Entry { key, _ := trunk.RepositoryKey(repository) @@ -72,7 +72,7 @@ func TestTheViewDerivesHowFarEachRepositoryReached(t *testing.T) { // web changed at A (fpOld) and at C (fpNew); mail first appeared at A. The source named A, B, then C as web's head. world := func(t *testing.T) []Entry { cur, _ := meshdeliveryclient.DecodeSource(src(12, commitC).Value) - current := Entry{Bucket: SourceBucket, Key: meshdeliveryclient.SourceMainKey, Revision: 12, At: t0} + current := Entry{Bucket: SourceBucket, Key: trunkSourceKey, Revision: 12, At: t0} current.Value, _ = meshdeliveryclient.EncodeSource(cur) return []Entry{src(10, commitA), src(11, commitB), src(12, commitC), current, mod("web", fpOld, commitA, 10, true), modIn(fpNew, commitC, 12, in("t1", "mesh-sdk/go"), true), mod("mail", fpMail, commitA, 0, true), diff --git a/modules/mesh-delivery-view/internal/view/tiers.go b/modules/mesh-delivery-view/internal/view/tiers.go index a579c044..1d720ef5 100644 --- a/modules/mesh-delivery-view/internal/view/tiers.go +++ b/modules/mesh-delivery-view/internal/view/tiers.go @@ -13,7 +13,6 @@ import ( "git.novox.be/novox/mesh-contracts/health-and-repair/go/clients/meshwatchdogsclient" "git.novox.be/novox/mesh-contracts/provisioning/go/clients/meshprovisionerclient" "git.novox.be/novox/mesh-delivery/contracts/go/clients/meshcatalogclient" - "git.novox.be/novox/mesh-delivery/contracts/go/clients/meshdeliveryclient" "git.novox.be/novox/mesh-delivery/contracts/go/clients/meshwantsclient" "git.novox.be/novox/mesh-delivery/contracts/go/seats/meshcatalog" "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-builder/builds" @@ -83,7 +82,7 @@ func (r Reader) Tiers(revision uint64) (Tiers, error) { if r.SourceState.GetEntry == nil { return Tiers{}, fmt.Errorf("the gate's source cannot be read here: name the source revision") } - _, rev, err := r.SourceState.GetEntry(meshdeliveryclient.SourceMainKey) + _, rev, err := r.SourceState.GetEntry(trunkSourceKey) if err != nil { return Tiers{}, fmt.Errorf("the gate's source: %w", err) } diff --git a/modules/mesh-delivery-view/internal/view/view.go b/modules/mesh-delivery-view/internal/view/view.go index 35f1c077..49db0b8e 100644 --- a/modules/mesh-delivery-view/internal/view/view.go +++ b/modules/mesh-delivery-view/internal/view/view.go @@ -621,7 +621,7 @@ func (m *Model) apply(e Entry) { delete(m.taken, e.Key) } case SourceBucket: - if e.Key != meshdeliveryclient.SourceMainKey { + if e.Key != trunkSourceKey { return } if e.Deleted { diff --git a/modules/mesh-delivery-view/internal/view/wants_test.go b/modules/mesh-delivery-view/internal/view/wants_test.go index 6599683e..23a84f33 100644 --- a/modules/mesh-delivery-view/internal/view/wants_test.go +++ b/modules/mesh-delivery-view/internal/view/wants_test.go @@ -61,11 +61,11 @@ func moduleE(t *testing.T, module, fp string) Entry { func sourceE(t *testing.T, revision uint64) Entry { t.Helper() - v, err := meshdeliveryclient.EncodeSource(meshdelivery.SourceByMainValue{Group: "g", Heads: map[string]string{"novox/web": strings.Repeat("1", 40)}}) + v, err := meshdeliveryclient.EncodeSource(meshdelivery.SourceByBranchValue{Group: "g", Heads: map[string]string{"novox/web": strings.Repeat("1", 40)}}) if err != nil { t.Fatal(err) } - return Entry{Bucket: SourceBucket, Key: meshdeliveryclient.SourceMainKey, Value: v, At: t0, Revision: revision} + return Entry{Bucket: SourceBucket, Key: trunkSourceKey, Value: v, At: t0, Revision: revision} } // 1906/D7 AC7: every want is a row beside the have of its node, with where it stands. From e7bd54e67bda305fb2cb5c3513f25dccfe4b79a1 Mon Sep 17 00:00:00 2001 From: jochen Date: Wed, 11 Nov 2026 11:51:07 +0100 Subject: [PATCH 06/10] IX19: the indexer reads every contracts module's trunk definitions at its start before judging, waits for the seat git at most two minutes, says view-waits past five and judges again once it answers MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Taken from the join base (6ce86c96, ff6e74f9) onto the build branch. Builds: design 47 G §198, §248, §316, §255, §57 (tests 54, 57) Claude-Session: https://claude.ai/code/session_01JG73KEq4zoDfM9WtYpcQgZ --- .../cmd/mesh-delivery-indexer/main.go | 28 ++- .../cmd/mesh-delivery-indexer/views.go | 25 +++ .../internal/index/control_records_test.go | 59 +++++++ .../internal/index/seatdefs.go | 46 +++++ .../internal/index/startread.go | 162 ++++++++++++++++++ 5 files changed, 315 insertions(+), 5 deletions(-) create mode 100644 modules/mesh-delivery-indexer/internal/index/startread.go diff --git a/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/main.go b/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/main.go index 01e3af36..8bb6284f 100644 --- a/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/main.go +++ b/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/main.go @@ -591,6 +591,14 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro if err := ix.Reconcile(ctx); err != nil { logger.Printf("dropping the modules the trunk does not hold at start: %v", err) } + // Every contracts module's trunk definitions read before any manifest is judged and before the forge bridge's stream is + // followed (design 47, the indexer reads every definition before it judges): a definition merged while it was down, or + // one no event announced, is never missing from the judging below. Once the state is bound, so on its first binding too. + // Through the seat git (milestone 47/G W4.15, design 47 *A part that cannot read the forge keeps what it has*): a seat that + // does not answer is waited for at most index.GitWaitBound before the judging below runs on the kept definitions, said in + // indexer_first_views and as view-waits past five minutes, and read again on the next forge message heard. + defs := gitWaits(ctx, ix, findings, process.AwaitBucket(ctx, js, conn, firstViewsBucket, "indexer_first_views does not say the wait for the seat git", logger.Printf), logger.Printf) + seatDefs.Ready(func(jetstream.KeyValue) { defs.Start(ctx) }) // Every manifest judged at the start, so a judge that changed with this program judges every one again (ADR 0578 // §1), once the verdicts state is there; each merge judges them again (index.ApplyMessage). verdicts.Ready(func(jetstream.KeyValue) { @@ -603,13 +611,22 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro }() // A pass that fails (a read the bus did not answer) is judged again at the next message or source revision the // indexer hears (index.kickJudge): no timer. - j, err := ix.JudgeAtStart(ctx) - if err != nil { - logger.Printf("judging the manifests at start: %v; judged again at the next message heard", err) + judge := func(fromKept bool) { + j, err := ix.JudgeAtStart(ctx) + if err != nil { + logger.Printf("judging the manifests at start (from kept definitions: %v): %v; judged again at the next message heard", fromKept, err) + return + } + logger.Printf("judged %d manifests at start (from kept definitions: %v): %d verdicts written, %d with problems, of which outgrown and kept accepted: %v, waiting for a judge that knows their fields: %v", + j.Judged, fromKept, j.Written, len(j.Refused), j.Outgrown, j.Waiting) + } + if !defs.Waiting() { + judge(false) return } - logger.Printf("judged %d manifests at start: %d verdicts written, %d with problems, of which outgrown and kept accepted: %v, waiting for a judge that knows their fields: %v", - j.Judged, j.Written, len(j.Refused), j.Outgrown, j.Waiting) + // The seat git did not answer: the judging waits for it at most index.GitWaitBound, off the start, so nothing else of + // the start is held behind it. + go func() { judge(defs.Ready(ctx)) }() }) // A build module says each time it wrote its toolchains key: its stream, and the indexer's durable consumer on it, are // made by the controller as the forge's are. Where the stream is not there yet the indexer says so and is started again @@ -736,6 +753,7 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro handle(ctx, m, ix, heard, logger) ix.KickFill(ctx) unreadAgain(ctx, ix, logger) + defs.Heard(ctx) }, logger.Printf) if err != nil { return err diff --git a/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/views.go b/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/views.go index 89d56bdf..a1afd067 100644 --- a/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/views.go +++ b/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/views.go @@ -3,6 +3,7 @@ package main import ( "context" "encoding/json" + "git.novox.be/novox/mesh-delivery/contracts/go/seats/git" "sync/atomic" "time" @@ -180,3 +181,27 @@ func followAssignments(ctx context.Context, hub *process.EventHub, cred process. } return f } + +// gitWaits is the indexer's read of the trunk definitions at its start through the seat git (index.DefinitionsAtStart): its +// wait said as a first view of the seat's (Owner git, State seat-definitions) in the indexer's first views, and its finding +// view-waits written past views.WaitsBound and cleared once the seat answers. +func gitWaits(ctx context.Context, ix *index.Indexer, findings, status *process.AwaitedBucket, logf func(string, ...any)) *index.DefinitionsAtStart { + finding := func(s views.Status, raised bool) { + w := sdkfindings.Writer(ctx, findings.KV) + f := views.ViewWaitsFinding(index.Module, s, time.Now()) + var err error + if raised { + err = w.Put(f) + } else { + err = w.Delete(f) + } + if err != nil { + logf("views: the finding view-waits of the seat git is not written: %v", err) + } + } + f := &views.Follower{Owner: git.Seat, Tool: "seat:" + git.Seat + "." + git.VerbTrees, State: index.SeatDefinitionsState, Logf: logf, + RaiseWaits: func(s views.Status) { finding(s, true) }, ClearWaits: func(s views.Status) { finding(s, false) }} + status.Ready(func(kv jetstream.KeyValue) { f.Said = views.KeepStatus(ctx, kv, logf) }) + return &index.DefinitionsAtStart{Read: ix.ReadTrunkDefinitions, Waits: f, Logf: logf, + Judge: func(ctx context.Context) error { _, err := ix.JudgeAtStart(ctx); return err }} +} diff --git a/modules/mesh-delivery-indexer/internal/index/control_records_test.go b/modules/mesh-delivery-indexer/internal/index/control_records_test.go index 66f7948c..337fb881 100644 --- a/modules/mesh-delivery-indexer/internal/index/control_records_test.go +++ b/modules/mesh-delivery-indexer/internal/index/control_records_test.go @@ -125,6 +125,36 @@ func TestTheControlRecords(t *testing.T) { // and source, says seat-definition-changed only when a digest moved, keeps the trunk's entry over one that does not // parse (its finding), removes a pull request's entries when it closes, keeps a verdict's out-of-step files, and // raises generated-types-out-of-step from a trunk build end, cleared only by a clean one of the same module. + // Milestone 47/G W4.15, design 47 test 54: the seat git not answering at start holds the judging at most its bound, which + // then runs on the kept definitions with the wait said; the read once the forge is heard holds the view and judges again. + t.Run("W4.15 the start waits for the seat git, judges from kept definitions, and judges again once it answers", func(t *testing.T) { + answers := []error{fmt.Errorf("mesh-contracts: %w", errForgeRead), fmt.Errorf("again: %w", errForgeRead), nil} + var said []string + judged := 0 + d := &DefinitionsAtStart{Bound: time.Millisecond, Waits: waitsFake{&said}, + Read: func(context.Context) (int, error) { err := answers[0]; answers = answers[1:]; return 1, err }, + Judge: func(context.Context) error { judged++; return nil }} + ctx := context.Background() + d.Start(ctx) + if !d.Waiting() || len(said) != 1 || said[0] != "waits "+GitDefinitionsView { + t.Fatalf("waiting %v, said %q: want the wait for the seat git said once", d.Waiting(), said) + } + if !d.Ready(ctx) || judged != 0 { + t.Fatalf("the start past its bound: want it judged from the kept definitions, judged again %d", judged) + } + d.Heard(ctx) // the forge heard, the seat still not answering: the wait stands, said once + if !d.Waiting() || len(said) != 1 || judged != 0 { + t.Fatalf("a read still not answered: waiting %v, said %q, judged %d", d.Waiting(), said, judged) + } + d.Heard(ctx) + if d.Waiting() || len(said) != 2 || said[1] != "holds "+GitDefinitionsView || judged != 1 || d.Ready(ctx) { + t.Errorf("the seat answering: waiting %v, said %q, judged again %d: want the view held and every manifest judged again once", d.Waiting(), said, judged) + } + d.Heard(ctx) // nothing waits: no read, no judging + if judged != 1 || len(answers) != 0 { + t.Errorf("a forge message with nothing waiting read or judged again") + } + }) t.Run("seat definitions", func(t *testing.T) { const repo, m1, m2, m3, head = "novox/mesh-contracts", "m1", "m2", "m3", "h1" manifest := `{"module":"mesh-contracts","version":"1.0","build":{"artifacts":[{"name":"go","kind":"package","language":"go","paths":["go"],"series":"0.1","generated-from":"."}]}}` @@ -315,6 +345,29 @@ func TestTheControlRecords(t *testing.T) { t.Errorf("said %+v: only a trunk's definition is said (ADR 0636 §4)", c) } } + // Design 47 (the indexer reads every definition before it judges): at its start the indexer reads every contracts + // module's trunk at its repository's head in the source revision, so a definition merged with no event heard is kept + // before any manifest is judged; a repository the source names no head of is read at the module's own commit. + kept, _ = meshcatalogclient.EncodeManifest(meshcatalog.ManifestsByDigestValue{Module: "mesh-contracts", Manifest: manifest}) + mkey, _ = meshcatalogclient.ManifestsKey(meshcatalogclient.ManifestDigest([]byte(manifest))) + ix.Manifests.(*memStore).vals[mkey] = kept + mod, _ = meshcatalogclient.EncodeModule(meshcatalog.ModulesByModuleValue{Fp2: "src2:" + strings.Repeat("a", 64), Manifest: meshcatalogclient.ManifestDigest([]byte(manifest)), Repository: repo, Commit: m3, Path: meshcatalogclient.RootPath}) + ix.Modules.(*memStore).vals["mesh-contracts"] = mod + files["h3"] = map[string]string{"seats/git/seat.json": strings.Replace(git, "ADR 0111", "ADR 0500", 1)} + ix.source = sourceNow{rev: 7, heads: map[string]string{repo: "h3"}} + if n, err := ix.ReadTrunkDefinitions(ctx); err != nil || n != 1 { + t.Fatalf("read at start = %d, %v; want the one contracts module", n, err) + } + if e, _ := entry("git.trunk"); e.Commit != "h3" { + t.Errorf("git.trunk after the read at start = %+v, want the source's head h3", e) + } + ix.source = sourceNow{} + if _, err := ix.ReadTrunkDefinitions(ctx); err != nil { + t.Fatal(err) + } + if e, _ := entry("git.trunk"); e.Commit != m3 { + t.Errorf("git.trunk read with no source head = %+v, want the module's own commit", e) + } // ADR 0636 §5, D-I3, D-I5: a seat of the mesh's own contracts is taken only from its domain's contracts module; a // move there is a handover in either order of the two trunks; a second definer is refused with a finding naming both. @@ -636,3 +689,9 @@ func (m memSource) Keys() ([]string, error) { } func (m memSource) Get(key string) (*views.Entry, bool, error) { e, ok := m[key]; return e, ok, nil } + +// waitsFake records what the indexer says it waits for and holds. +type waitsFake struct{ said *[]string } + +func (w waitsFake) WaitFor(_ context.Context, view string) { *w.said = append(*w.said, "waits "+view) } +func (w waitsFake) Holds(view string) { *w.said = append(*w.said, "holds "+view) } diff --git a/modules/mesh-delivery-indexer/internal/index/seatdefs.go b/modules/mesh-delivery-indexer/internal/index/seatdefs.go index 87d3125e..488bb37f 100644 --- a/modules/mesh-delivery-indexer/internal/index/seatdefs.go +++ b/modules/mesh-delivery-indexer/internal/index/seatdefs.go @@ -13,12 +13,14 @@ package index // contracts module. import ( + "cmp" "context" "crypto/sha256" "encoding/hex" "encoding/json" "errors" "fmt" + "maps" "path" "slices" "strconv" @@ -30,6 +32,7 @@ import ( "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-builder/builds" "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-forge-bridge/forgemsg" "git.novox.be/novox/mesh-delivery/sdk/seatdefs" + "git.novox.be/novox/mesh-delivery/sdk/trunk" "git.novox.be/novox/mesh-sdk/go/keys" "git.novox.be/novox/mesh-sdk/go/seatdef" "git.novox.be/novox/mesh-sdk/go/views" @@ -278,6 +281,49 @@ func (ix *Indexer) SeatDefinitionsAgain(ctx context.Context, module string, pull return out } +// ReadTrunkDefinitions reads, at the start and before any manifest is judged, the seat definitions of every contracts +// module the trunk holds, at its repository's head in the source revision (at the module's own commit where the source +// names none), and keeps them as a merge does (design 47, *The indexer reads every definition before it judges*): a +// definition merged while the indexer was down, or one no event announced, is kept before a manifest claiming its seat +// is judged. It answers how many contracts modules were read; one the forge did not answer is in the error, and is read +// again at its next merged message or by indexer_seat_definitions_again. +func (ix *Indexer) ReadTrunkDefinitions(ctx context.Context) (int, error) { + if ix.SeatDefinitions == nil || ix.ReadFile == nil || ix.Modules == nil || ix.Manifests == nil { + return 0, nil + } + held, err := ix.heldModules(ctx) + if err != nil { + return 0, err + } + ix.mu.Lock() + heads := maps.Clone(ix.source.heads) + ix.mu.Unlock() + read := 0 + var errs []error + for _, name := range slices.Sorted(maps.Keys(held)) { + m := held[name] + manifest, _, err := ix.manifestRaw(ctx, m.Manifest) + if err != nil { + errs = append(errs, fmt.Errorf("%s's manifest: %w", name, err)) + continue + } + dir := cmp.Or(m.Path, meshcatalogclient.RootPath) + commit := m.Commit + for r, head := range heads { + if trunk.SameRepository(r, m.Repository) && head != "" { + commit = head + } + } + for _, c := range contractsModules(map[string]string{dir: manifest}) { + read++ + if err := ix.keepTrunk(ctx, m.Repository, commit, c); err != nil { + errs = append(errs, fmt.Errorf("%s at %.12s: %w", name, commit, err)) + } + } + } + return read, errors.Join(errs...) +} + // keepTrunk writes a contracts module's trunk definitions at a merge: each that parses where its digest moved (said), // each the module's index no longer names removed (said with no digest), and the module's unparsed finding written or // cleared. A definition that does not parse leaves the trunk's entry as it stands. diff --git a/modules/mesh-delivery-indexer/internal/index/startread.go b/modules/mesh-delivery-indexer/internal/index/startread.go new file mode 100644 index 00000000..feea611a --- /dev/null +++ b/modules/mesh-delivery-indexer/internal/index/startread.go @@ -0,0 +1,162 @@ +package index + +import ( + "context" + "errors" + "sync" + "time" +) + +// GitWaitBound is how long the indexer waits, at its start, for the seat git to answer the read of every contracts module's +// trunk definitions before it judges from the definitions it kept (design 47, *A part that cannot read the forge keeps +// what it has, and says so*): longer than a forge's restart, short enough that a start is not held behind a forge down +// for good. +const GitWaitBound = 2 * time.Minute + +// GitDefinitionsView is what the indexer says it waits for in its first views while the seat git does not answer. +const GitDefinitionsView = "trunk seat definitions" + +// ViewWaiter is where the indexer says it waits for the seat git and since when, and raises view-waits past its bound +// (mesh-sdk views.Follower's WaitFor and Holds). +type ViewWaiter interface { + WaitFor(ctx context.Context, view string) + Holds(view string) +} + +// DefinitionsAtStart is the indexer's read of the trunk definitions at its start through the seat git: the first read, +// a wait of at most Bound for the seat when it does not answer, the judging from the kept definitions after it, and the +// read again once the seat answers, which judges again every manifest judged from kept definitions. +type DefinitionsAtStart struct { + // Read reads every contracts module's trunk definitions (Indexer.ReadTrunkDefinitions). + Read func(ctx context.Context) (int, error) + // Judge judges every manifest again, as at the start (Indexer.JudgeAtStart). + Judge func(ctx context.Context) error + Waits ViewWaiter + Bound time.Duration + Logf func(string, ...any) + + mu sync.Mutex + answered chan struct{} // closed once a read is answered + waiting bool // a read was not answered and no read since was + fromKept bool // the judging at start ran on kept definitions +} + +func (d *DefinitionsAtStart) init() { + if d.answered == nil { + d.answered = make(chan struct{}) + } +} + +func (d *DefinitionsAtStart) logf(format string, args ...any) { + if d.Logf != nil { + d.Logf(format, args...) + } +} + +// Start reads the trunk definitions once; a seat that does not answer is said as a wait (Waits.WaitFor) and read again on +// the next sign that the forge runs (Heard). A read answered with other errors (a manifest it cannot read) is answered. +func (d *DefinitionsAtStart) Start(ctx context.Context) { + d.mu.Lock() + d.init() + d.mu.Unlock() + d.read(ctx, "at start") +} + +func (d *DefinitionsAtStart) read(ctx context.Context, when string) bool { + n, err := d.Read(ctx) + if errors.Is(err, errForgeRead) { + d.mu.Lock() + first := !d.waiting + d.waiting = true + d.mu.Unlock() + if first { + d.logf("the seat git did not answer the read of %d contracts modules' trunk definitions %s: %v; the indexer judges from the definitions it kept after %s, and reads them again once the seat answers", n, when, err, d.bound()) + if d.Waits != nil { + d.Waits.WaitFor(ctx, GitDefinitionsView) + } + } + return false + } + if err != nil { + d.logf("reading the seat definitions of %d contracts modules' trunks %s: %v; each is read again at its next merged message or by indexer_seat_definitions_again", n, when, err) + } else { + d.logf("read the seat definitions of %d contracts modules' trunks %s", n, when) + } + d.mu.Lock() + was := d.waiting + d.waiting = false + select { + case <-d.answered: + default: + close(d.answered) + } + again := d.fromKept + d.fromKept = false + d.mu.Unlock() + if was && d.Waits != nil { + d.Waits.Holds(GitDefinitionsView) + } + if again && d.Judge != nil { + if err := d.Judge(ctx); err != nil { + d.logf("judging again the manifests judged from kept definitions: %v; judged again at the next message heard", err) + } else { + d.logf("judged again every manifest judged from kept definitions, now that the seat git answers") + } + } + return true +} + +func (d *DefinitionsAtStart) bound() time.Duration { + if d.Bound > 0 { + return d.Bound + } + return GitWaitBound +} + +// Ready waits, while the seat git does not answer the read at start, until a read is answered, or Bound passed (no timer: a context's bound), and says whether the +// judging that follows runs on the definitions the indexer kept; it marks it so, and the read that answers later judges +// again (Read's answer). +func (d *DefinitionsAtStart) Ready(ctx context.Context) (fromKept bool) { + d.mu.Lock() + d.init() + answered, waiting := d.answered, d.waiting + d.mu.Unlock() + if !waiting { + return false // read, or not read at all (the seat definitions' state not bound yet): nothing to wait for + } + wait, cancel := context.WithTimeout(ctx, d.bound()) + defer cancel() + select { + case <-answered: + return false + case <-wait.Done(): + } + d.mu.Lock() + defer d.mu.Unlock() + select { + case <-d.answered: + return false + default: + } + d.fromKept = true + d.logf("the seat git has not answered within %s: the indexer judges from the definitions it kept", d.bound()) + return true +} + +// Heard is a sign that the forge runs (a message the forge bridge relays): while the indexer waits for the seat git, it +// reads the trunk definitions again. +func (d *DefinitionsAtStart) Heard(ctx context.Context) { + d.mu.Lock() + waiting := d.waiting + d.mu.Unlock() + if waiting { + d.read(ctx, "once the forge was heard") + } +} + +// Waiting says the indexer waits for the seat git. +func (d *DefinitionsAtStart) Waiting() bool { + d.mu.Lock() + defer d.mu.Unlock() + return d.waiting +} From 3b31e6df05087fcdb2e854d38dd4dc90ddbfe5fd Mon Sep 17 00:00:00 2001 From: jochen Date: Wed, 11 Nov 2026 11:53:07 +0100 Subject: [PATCH 07/10] AS1: a want follows a passed pipeline, by compare-and-set per module and watched branch MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The assigner hears pipeline-ended and group-ended, keeps the newest end per module and branch in the state written (tool assigner_written), and writes a want only from an end whose required steps passed. Builds: design 47 G §29, §35, §102, §120, §291, §325, §352, §85 Claude-Session: https://claude.ai/code/session_01JG73KEq4zoDfM9WtYpcQgZ --- .../cmd/mesh-delivery-assigner/main.go | 17 +- .../cmd/mesh-delivery-assigner/pass.go | 14 + .../cmd/mesh-delivery-assigner/pipelines.go | 327 ++++++++++++++++++ .../mesh-delivery-assigner/pipelines_test.go | 174 ++++++++++ .../cmd/mesh-delivery-assigner/tools.go | 1 + .../cmd/mesh-delivery-assigner/world.go | 14 + .../mesh-delivery-assigner/manifest_test.go | 6 +- modules/mesh-delivery-assigner/module.json | 8 +- 8 files changed, 554 insertions(+), 7 deletions(-) create mode 100644 modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/pipelines.go create mode 100644 modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/pipelines_test.go diff --git a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/main.go b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/main.go index 3ec09822..4f040bed 100644 --- a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/main.go +++ b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/main.go @@ -79,7 +79,7 @@ var readFamilies = []string{buckets.ComposedRead} var ownBuckets = []string{buckets.Asks, buckets.Differences, buckets.Puts, buckets.Wants, plc.AssignmentsBucket, plc.ProposalsBucket, plc.CandidatesBucket, plc.PulsesBucket, assign.FindingsBucket, assign.PoliciesBucket, plc.SeatHoldersBucket, - engineReadsBucket, retiredBucket, copiesBucket} + engineReadsBucket, retiredBucket, copiesBucket, writtenBucket} // engineReadsState is the assigner's state of what each node's engine states it reads (ADR 0618 §3), one key ``, as // the verb manifest-reads last answered it or the fallback read it: what assigner_engine_reads lists and asks again. @@ -88,6 +88,9 @@ const engineReadsState = "engine-reads" // engineReadsBucket is that state's bucket. const engineReadsBucket = assignment.Module + "_" + engineReadsState +// writtenBucket is the bucket of the state written (pipelines.go): the pipeline ends the wants follow. +const writtenBucket = assignment.Module + "_" + writtenState + // seatBuckets are the seat's states, new with it: until the controller has made them from the manifest and the bus grants // them, the assigner starts without them and binds each once the bus says it was created (issue 1790's kind). var seatBuckets = []string{plc.AssignmentsBucket, plc.ProposalsBucket, plc.CandidatesBucket, @@ -115,6 +118,7 @@ var keptStates = []announce.State{ {Name: engineReadsState, Readable: true}, {Name: retiredState}, // read only by the copy into assignments (migrate.go) {Name: copiesState, Readable: true}, + {Name: writtenState, Readable: true}, } // kvSet is every bucket the assigner opened, by name. @@ -197,7 +201,7 @@ func open(ctx context.Context, js process.BucketOpener, logf func(string, ...any func startsWithout(name string) bool { switch name { case plc.SeatHoldersBucket, buckets.Provisions, assign.PoliciesBucket, assign.FindingsBucket, - buckets.Puts, buckets.Source, buckets.Merges, engineReadsBucket, retiredBucket: + buckets.Puts, buckets.Source, buckets.Merges, engineReadsBucket, retiredBucket, writtenBucket: return true } return slices.Contains(seatBuckets, name) || name == buckets.Grants @@ -297,6 +301,7 @@ var watched = []string{ buckets.Source, buckets.Merges, engineReadsBucket, + writtenBucket, } // follow watches every bucket a pass reads and makes a pass at start, once every watch handed what stands, and after @@ -485,6 +490,14 @@ func follow(ctx context.Context, js jetstream.JetStream, nc process.Subscriber, logger.Printf("%s is not heard, so an ask that ends at the operator channel without an answer is not asked again until the assigner starts again: %v", askEndedSubject, err) } readEndedAtBind(ctx, js, hub, kvs[buckets.Asks], staleOfWorld(w), bus, callWait{heard: func() bool { return w.seatHeard(oc.Seat) }, findings: findings}, logger.Printf) + if kv := kvs[writtenBucket]; kv != nil { + // The runner's pipeline-ended and group-ended (design 47 G §29): kept in the state written, which the wants follow + // once the first registrations are written (§95, §353). + p.heard = newPipelineEnds(kvEnds{ctx, kv}, wake, logger.Printf) + if err := hearEnds(hub, p.heard); err != nil { + logger.Printf("a pipeline's end is not heard, so no want follows a pipeline until the assigner starts again: %v", err) + } + } hearReadsChanged(ctx, js, hub, reads, findings, gate.eventsLive, logger.Printf) } if bus != nil && kvs[buckets.Puts] != nil { diff --git a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/pass.go b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/pass.go index 477ec9c2..69d369af 100644 --- a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/pass.go +++ b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/pass.go @@ -58,6 +58,10 @@ type passer struct { // neither is, by node and module, each module on neither list whose condition this process raised (issue 1945). Nil until the // first pass after a start, which clears what a previous process may have raised (startClears). neither map[string]neitherOne + // ends are the pipeline ends the wants follow (pipelines.go); nil is the trunk-only want, kept until the first registrations. + ends *pipelineEnds + // heard is the same follower while the wants do not follow it yet (ends is nil). + heard *pipelineEnds // acts are the puts' acts (ADR 0546); nil: the puts are not taken further (no bus to call, or no puts bucket). acts *putActs // noWant is, by node and module, the reason last said why the node gets no want for the module (issue 1945): a reason is @@ -267,9 +271,16 @@ func (p *passer) wants(ctx context.Context) { moduleGraph := p.moduleGraph() served := p.served(ctx, get, moduleGraph, nodes) offering := p.offering(ctx, get, manifests, nodes) // bus state (ADR 0614 §5) + assignedBranches := p.w.placedBranches() for name := range names { fp := trunk[name].Fp2 name := name + // A want follows a passed pipeline: a module whose fingerprint no pipeline passed at on every branch its placements + // chose, and whose wants do not name it already, gets none (design 47 G §29, §35). + branches := p.ends.branchesOf(name, assignedBranches) + if fp != "" && !p.ends.mayWant(name, fp, branches, wanted[name]) { + continue + } m := assign.Module{Name: name, FP: fp, Manifest: manifests[name], Needs: trunk[name].Needs, Chosen: chosenOf(chosen, name), Provisions: provisionsOf(name), Answering: answering(manifests, nodes), Offered: offered(manifests), Nodes: nodesOf[name], Now: time.Now(), Puts: puts[name], FormNeeds: p.w.buildNeeds(name, fp), Self: func(node string) bool { return placed[node] }, Composed: func(node, fp string) (assign.ComposedDigest, bool) { @@ -359,6 +370,9 @@ func (p *passer) wants(ctx context.Context) { want := want p.put(ctx, buckets.Wants, key, func() ([]byte, error) { return mwc.EncodeWant(want) }) } + if len(out.Wants) > 0 { + p.ends.wrote(name, fp, branches) + } } } diff --git a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/pipelines.go b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/pipelines.go new file mode 100644 index 00000000..743125a5 --- /dev/null +++ b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/pipelines.go @@ -0,0 +1,327 @@ +package main + +import ( + "context" + "encoding/json" + "errors" + "fmt" + "sort" + "sync" + + "github.com/nats-io/nats.go/jetstream" + + mprc "git.novox.be/novox/mesh-delivery/contracts/go/clients/meshpipelinerunnerclient" + mwc "git.novox.be/novox/mesh-delivery/contracts/go/clients/meshwantsclient" + mpr "git.novox.be/novox/mesh-delivery/contracts/go/seats/meshpipelinerunner" + stdio "git.novox.be/novox/mesh-sdk/go" + "git.novox.be/novox/mesh-sdk/go/keys" + "git.novox.be/novox/mesh-sdk/go/process" +) + +// The wants follow a passed pipeline (design 47 G §29, §35, §325, §352): the runner says each pipeline's end as +// pipeline-ended, and a group's as group-ended. The assigner keeps, in its old state `written`, one key per module and +// watched branch: the source revision and end count of the newest end it heard for it, and whether the wants were written +// from it. A newer end replaces it by compare-and-set; an older one, or the same one again, writes nothing. + +// writtenState is the assigner's state of the pipeline ends its wants follow; writtenTool reads it. +const ( + writtenState = "written" + writtenTool = "assigner_written" +) + +// pipelineEndedSubject and groupEndedSubject are the runner's events the assigner consumes. +const ( + pipelineEndedSubject = "mesh.seat." + mpr.Seat + ".event." + mpr.EventPipelineEnded + ".>" + groupEndedSubject = "mesh.seat." + mpr.Seat + ".event." + mpr.EventGroupEnded + ".>" +) + +// End is one value of the state written: the newest end of a module's pipeline on a watched branch. +type End struct { + Source int64 `json:"source"` + EndCount int64 `json:"end-count"` + Fp string `json:"fp"` + Commit string `json:"commit,omitempty"` + // Passed: the pipeline ended passed, every required step passed (and, for a group, the group ended passed). + Passed bool `json:"passed"` + // Written: the wants were written from this end. + Written bool `json:"written,omitempty"` + Why string `json:"why,omitempty"` +} + +// endKey is the key of a module on a watched branch, each name escaped as a key's name is. +func endKey(module, branch string) string { + return keys.EscapeName(module) + "." + keys.EscapeName(branch) +} + +// newer says whether an end of (source, count) replaces the one kept: a later source revision, or the same one ended again. +func (e End) newer(source, count int64) bool { + return source > e.Source || (source == e.Source && count > e.EndCount) +} + +// requiredPassed says whether no required step of a pipeline stands failed, unasked or unended (design 47 G §30: a failed +// optional step holds nothing). +func requiredPassed(p mpr.PipelineEnded) bool { + for _, s := range p.Steps { + if s.Status != mpr.StepResultStatusRequired { + continue + } + switch s.Outcome { + case mpr.StepResultOutcomeFailed, mpr.StepResultOutcomeCouldNotRun, mpr.StepResultOutcomeDeclined, mpr.StepResultOutcomeNotAsked, + mpr.StepResultOutcomeCancelled, mpr.StepResultOutcomeSuperseded, mpr.StepResultOutcomeDidNotBuild, + mpr.StepResultOutcomeRunning, mpr.StepResultOutcomeWaiting, mpr.StepResultOutcomeWaitingForABuildNode: + return false + } + } + return p.Outcome == mpr.PipelineEndedOutcomePassed +} + +// endStore is the kept state written: the bus bucket, or a test's. +type endStore interface { + // get is a key's value and revision; none for a key not there. + get(key string) (value []byte, revision uint64, err error) + // swap writes value at key if the key stands at revision (0: if it does not stand); errSwapped when it does not. + swap(key string, value []byte, revision uint64) error +} + +var errSwapped = errors.New("the key changed") + +// kvEnds is endStore over a bus bucket. +type kvEnds struct { + ctx context.Context + kv jetstream.KeyValue +} + +func (k kvEnds) get(key string) ([]byte, uint64, error) { + e, err := k.kv.Get(k.ctx, key) + if errors.Is(err, jetstream.ErrKeyNotFound) { + return nil, 0, nil + } + if err != nil { + return nil, 0, err + } + return e.Value(), e.Revision(), nil +} + +func (k kvEnds) swap(key string, value []byte, revision uint64) error { + var err error + if revision == 0 { + _, err = k.kv.Create(k.ctx, key, value) + } else { + _, err = k.kv.Update(k.ctx, key, value, revision) + } + if errors.Is(err, jetstream.ErrKeyExists) || (err != nil && isWrongSequence(err)) { + return errSwapped + } + return err +} + +func isWrongSequence(err error) bool { + var ae *jetstream.APIError + return errors.As(err, &ae) && ae.ErrorCode == jetstream.JSErrCodeStreamWrongLastSequence +} + +// pipelineEnds follows the runner's ends. A nil *pipelineEnds is the trunk-only want, kept until the first registrations +// (design 47 G §95, §353). +type pipelineEnds struct { + st endStore + wake func() + logf func(string, ...any) + mu sync.Mutex + // firsts is, by module, the first watched branch of its registration, the branch of a placement that names none (design + // 47 G §14). + firsts map[string]string +} + +func newPipelineEnds(st endStore, wake func(), logf func(string, ...any)) *pipelineEnds { + return &pipelineEnds{st: st, wake: wake, logf: logf, firsts: map[string]string{}} +} + +// take keeps one pipeline's end if it is newer than the one kept for its module and branch, by compare-and-set. +func (e *pipelineEnds) take(p mpr.PipelineEnded, passed bool) error { + key := endKey(p.Module, p.Branch) + for { + raw, rev, err := e.st.get(key) + if err != nil { + return err + } + if raw != nil { + var kept End + if err := json.Unmarshal(raw, &kept); err == nil && !kept.newer(p.Source, p.EndCount) { + return nil // an older end, or this one again + } + } + end := End{Source: p.Source, EndCount: p.EndCount, Fp: p.Fp, Commit: p.Commit, Passed: passed, Why: p.Why} + value, err := json.Marshal(end) + if err != nil { + return err + } + if err := e.st.swap(key, value, rev); errors.Is(err, errSwapped) { + continue + } else if err != nil { + return err + } + e.wake() + return nil + } +} + +// markWritten says the wants of module on branch were written from the end it keeps, if that is still the end of fp. +func (e *pipelineEnds) markWritten(module, branch, fp string) { + key := endKey(module, branch) + for { + raw, rev, err := e.st.get(key) + var kept End + if err != nil || raw == nil || json.Unmarshal(raw, &kept) != nil || kept.Fp != fp || kept.Written { + return + } + kept.Written = true + value, _ := json.Marshal(kept) + if err := e.st.swap(key, value, rev); errors.Is(err, errSwapped) { + continue + } else if err != nil { + e.logf("the end of %s on %s is not marked written: %v", module, branch, err) + } + return + } +} + +// passedAt says whether the newest end kept for module on branch passed at fp. +func (e *pipelineEnds) passedAt(module, branch, fp string) bool { + raw, _, err := e.st.get(endKey(module, branch)) + var kept End + return err == nil && raw != nil && json.Unmarshal(raw, &kept) == nil && kept.Passed && kept.Fp == fp +} + +// branchOf is the watched branch a placement names, else the first of the module's registration, else the trunk's. +func (e *pipelineEnds) branchOf(module, placed string) string { + if placed != "" { + return placed + } + e.mu.Lock() + defer e.mu.Unlock() + if b := e.firsts[module]; b != "" { + return b + } + return trunkBranch +} + +// hearPipeline takes a pipeline-ended. A pipeline of a group writes nothing on its own end, and a run on request delivers +// nothing (design 47 G §29, §64). +func (e *pipelineEnds) hearPipeline(m jetstream.Msg) { + p, err := mprc.PipelineEnded(m.Data()) + switch { + case err != nil: + e.logf("%s is not taken: %v", m.Subject(), err) + case p.Group != "" || p.Run != "": + default: + if err := e.take(p, requiredPassed(p)); err != nil { + e.logf("%s: %v; it is heard again", m.Subject(), err) + _ = m.Nak() + return + } + } + _ = m.Ack() +} + +// hearGroup takes a group-ended: each member's end, passed only if the group passed. +func (e *pipelineEnds) hearGroup(m jetstream.Msg) { + g, err := mprc.GroupEnded(m.Data()) + if err != nil { + e.logf("%s is not taken: %v", m.Subject(), err) + _ = m.Ack() + return + } + for _, p := range g.Pipelines { + if err := e.take(p, g.Outcome == mpr.GroupEndedOutcomePassed && requiredPassed(p)); err != nil { + e.logf("%s: %v; it is heard again", m.Subject(), err) + _ = m.Nak() + return + } + } + _ = m.Ack() +} + +// hearEnds registers the two events on the assigner's hub. +func hearEnds(hub *process.EventHub, e *pipelineEnds) error { + if err := hub.Handle(pipelineEndedSubject, e.hearPipeline); err != nil { + return fmt.Errorf("%s: %w", pipelineEndedSubject, err) + } + if err := hub.Handle(groupEndedSubject, e.hearGroup); err != nil { + return fmt.Errorf("%s: %w", groupEndedSubject, err) + } + return nil +} + +// writtenTools is the tool assigner_written: every key of the state written, with its end. +func writtenTools(st store) []stdio.Tool { + return []stdio.Tool{{Name: writtenTool, Description: "the pipeline ends the assigner's wants follow: per module and watched branch, the source revision and end count of the newest end heard, whether it passed and whether the wants were written from it.", + Run: func(map[string]any) (any, error) { + ks, err := st.keys(writtenState) + if err != nil { + return nil, err + } + sort.Strings(ks) + out := map[string]End{} + for _, k := range ks { + raw, _, ok, err := st.get(writtenState, k) + if err != nil || !ok { + continue + } + var e End + if json.Unmarshal(raw, &e) == nil { + out[k] = e + } + } + return out, nil + }}} +} + +// branchesOf is the watched branches the placements of module chose, each once; the default branch when none names one. +func (e *pipelineEnds) branchesOf(module string, placed map[string]map[string]string) []string { + if e == nil { + return nil + } + seen := map[string]bool{} + var out []string + for _, b := range placed[module] { + b = e.branchOf(module, b) + if !seen[b] { + seen[b] = true + out = append(out, b) + } + } + if len(out) == 0 { + out = append(out, e.branchOf(module, "")) + } + sort.Strings(out) + return out +} + +// mayWant says whether module's wants may name fp: a pipeline passed at fp on every branch, or the wants name fp already +// (a node placed later, a module taken away). +func (e *pipelineEnds) mayWant(module, fp string, branches []string, wanted map[string]mwc.Want) bool { + if e == nil { + return true + } + for _, w := range wanted { + if w.Fp == fp { + return true + } + } + for _, b := range branches { + if !e.passedAt(module, b, fp) { + return false + } + } + return true +} + +// wrote marks the end the wants of module were written from. +func (e *pipelineEnds) wrote(module, fp string, branches []string) { + if e == nil { + return + } + for _, b := range branches { + e.markWritten(module, b, fp) + } +} diff --git a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/pipelines_test.go b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/pipelines_test.go new file mode 100644 index 00000000..17597d49 --- /dev/null +++ b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/pipelines_test.go @@ -0,0 +1,174 @@ +package main + +import ( + "context" + "encoding/json" + "strings" + "testing" + "time" + + "git.novox.be/novox/mesh-delivery/contracts/go/clients/meshcatalogclient" + mwc "git.novox.be/novox/mesh-delivery/contracts/go/clients/meshwantsclient" + "git.novox.be/novox/mesh-delivery/contracts/go/seats/meshcatalog" + mpr "git.novox.be/novox/mesh-delivery/contracts/go/seats/meshpipelinerunner" + + oc "git.novox.be/novox/mesh-contracts/operator-and-conversation/go/seats/operatorchannel" + + "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-builder/builds" + "git.novox.be/novox/mesh-sdk/go/catalogue" + + "git.novox.be/novox/mesh-delivery/modules/mesh-grants/grantsstate" + + "git.novox.be/novox/mesh-contracts/go/clients/nodeengineclient" + "git.novox.be/novox/mesh-contracts/go/seats/nodeengine" + + "mesh-delivery-assigner/internal/buckets" +) + +// memEnds is endStore over a map, with revisions. +type memEnds struct { + vals map[string][]byte + revs map[string]uint64 +} + +func newMemEnds() *memEnds { return &memEnds{vals: map[string][]byte{}, revs: map[string]uint64{}} } + +func (m *memEnds) get(key string) ([]byte, uint64, error) { return m.vals[key], m.revs[key], nil } +func (m *memEnds) swap(key string, v []byte, rev uint64) error { + if m.revs[key] != rev { + return errSwapped + } + m.vals[key], m.revs[key] = v, rev+1 + return nil +} + +func pipe(module, branch string, source, count int64, fp, outcome string, steps ...mpr.StepResult) mpr.PipelineEnded { + return mpr.PipelineEnded{Module: module, Branch: branch, Source: source, EndCount: count, Fp: fp, Outcome: outcome, Steps: steps, + Pipeline: branch + "." + module, Commit: "c"} +} + +func step(status, outcome string) mpr.StepResult { + return mpr.StepResult{Step: "unit", Status: status, Outcome: outcome} +} + +// Design 47 G §29, §35, §325, §352 (tests 66, 67, 89): a want follows a passed pipeline, by compare-and-set per module and branch. +func TestAWantFollowsAPassedPipeline(t *testing.T) { + const fp1, fp2 = "fp1", "fp2" + pass, fail := mpr.PipelineEndedOutcomePassed, mpr.PipelineEndedOutcomeFailed + req, opt := mpr.StepResultStatusRequired, mpr.StepResultStatusOptional + for _, c := range []struct { + name string + ends []mpr.PipelineEnded + group *mpr.GroupEnded + module string + branch string + fp string + may bool + }{ + {"a failed required step writes no want", []mpr.PipelineEnded{pipe("web", "main", 5, 1, fp1, fail, step(req, "failed"))}, nil, "web", "main", fp1, false}, + {"a pipeline passed writes", []mpr.PipelineEnded{pipe("web", "main", 5, 1, fp1, pass, step(req, "passed"))}, nil, "web", "main", fp1, true}, + {"a failed optional step holds nothing", []mpr.PipelineEnded{pipe("web", "main", 5, 1, fp1, pass, step(req, "passed"), step(opt, "failed"))}, nil, "web", "main", fp1, true}, + {"a pipeline passed at another fingerprint", []mpr.PipelineEnded{pipe("web", "main", 5, 1, fp1, pass)}, nil, "web", "main", fp2, false}, + {"an older revision ending after a newer one writes nothing", []mpr.PipelineEnded{pipe("web", "main", 6, 1, fp2, fail), pipe("web", "main", 5, 1, fp1, pass)}, nil, "web", "main", fp1, false}, + {"another branch writes its own", []mpr.PipelineEnded{pipe("web", "main", 6, 1, fp2, fail), pipe("web", "beta", 5, 1, fp1, pass)}, nil, "web", "beta", fp1, true}, + {"a later end of the same revision writes", []mpr.PipelineEnded{pipe("web", "main", 5, 1, fp1, fail, step(req, "failed")), pipe("web", "main", 5, 2, fp1, pass, step(req, "passed"))}, nil, "web", "main", fp1, true}, + {"the same end again after a later one keeps the later", []mpr.PipelineEnded{pipe("web", "main", 5, 2, fp1, pass), pipe("web", "main", 5, 1, fp1, fail)}, nil, "web", "main", fp1, true}, + {"a member's own end writes nothing", []mpr.PipelineEnded{func() mpr.PipelineEnded { p := pipe("web", "main", 5, 1, fp1, pass); p.Group = "g"; return p }()}, nil, "web", "main", fp1, false}, + {"a group ended passed writes its members", nil, &mpr.GroupEnded{Branch: "main", Group: "g", Outcome: mpr.GroupEndedOutcomePassed, Pipelines: []mpr.PipelineEnded{pipe("web", "main", 5, 1, fp1, pass)}}, "web", "main", fp1, true}, + {"a group ended failed writes none", nil, &mpr.GroupEnded{Branch: "main", Group: "g", Outcome: mpr.GroupEndedOutcomeFailed, Pipelines: []mpr.PipelineEnded{pipe("web", "main", 5, 1, fp1, pass)}}, "web", "main", fp1, false}, + } { + woken := 0 + e := newPipelineEnds(newMemEnds(), func() { woken++ }, t.Logf) + for _, p := range c.ends { + if p.Group != "" { + continue // heard and ignored, as hearPipeline does + } + if err := e.take(p, requiredPassed(p)); err != nil { + t.Fatal(err) + } + } + if g := c.group; g != nil { + for _, p := range g.Pipelines { + if err := e.take(p, g.Outcome == mpr.GroupEndedOutcomePassed && requiredPassed(p)); err != nil { + t.Fatal(err) + } + } + } + if got := e.mayWant(c.module, c.fp, []string{c.branch}, nil); got != c.may { + t.Errorf("%s: may want = %v, want %v", c.name, got, c.may) + } + } + // A want that names the fingerprint already is not held by a later failure (a node placed afterwards gets it). + e := newPipelineEnds(newMemEnds(), func() {}, t.Logf) + if !e.mayWant("web", fp1, []string{"main"}, map[string]mwc.Want{"g14": {Fp: fp1}}) { + t.Error("a want naming the fingerprint was held") + } + // markWritten keeps the end and says it was written from; a newer end replaces it unwritten. + st := newMemEnds() + e = newPipelineEnds(st, func() {}, t.Logf) + _ = e.take(pipe("web", "main", 5, 1, fp1, pass), true) + e.wrote("web", fp1, []string{"main"}) + var end End + _ = json.Unmarshal(st.vals[endKey("web", "main")], &end) + if !end.Written || end.Source != 5 { + t.Errorf("end = %+v, want written at source 5", end) + } + _ = e.take(pipe("web", "main", 6, 1, fp2, pass), true) + end = End{} + _ = json.Unmarshal(st.vals[endKey("web", "main")], &end) + if end.Written || end.Source != 6 { + t.Errorf("a newer end = %+v, want unwritten at source 6", end) + } +} + +// Design 47 G §29: the wants of a pass follow a pipeline: with the follower on, a module whose fingerprint no pipeline passed at +// gets no want and one a pipeline passed at gets it; with it off (nil, the trunk-only want) the pass is as before. +func TestAPassWritesNoWantWithoutAPassedPipeline(t *testing.T) { + fp := meshcatalogclient.SourcePrefix2 + strings.Repeat("a", 64) + for _, c := range []struct { + name string + follow bool + passed bool + want bool + }{{"the trunk-only want", false, false, true}, {"no pipeline passed", true, false, false}, {"a pipeline passed", true, true, true}} { + wantsKV := &memKV{values: map[string][]byte{}} + p := newPasser(kvSet{buckets.Wants: wantsKV, buckets.Asks: &memKV{values: map[string][]byte{}}}, time.Now()) + p.say = func(oc.Condition) error { return nil } + mdigest := strings.Repeat("e", 64) + mod, _ := meshcatalogclient.EncodeModule(meshcatalog.ModulesByModuleValue{Fp2: fp, Manifest: mdigest, Repository: "novox/web", Commit: strings.Repeat("1", 40), Path: meshcatalogclient.RootPath}) + p.w.put(buckets.Modules, "web", mod) + p.w.manifests[mdigest] = parsed{module: "web", manifest: catalogue.Manifest{Module: "web"}} + bkey, _ := builds.BuilderBuildKey("web", fp) + built, _ := builds.EncodeBuilderBuild(builds.BuilderBuild{State: builds.BuildDone, Manifest: mdigest}) + p.w.put(buckets.Builds, bkey, built) + gkey, _ := grantsstate.PartGrantKey("web", fp) + grant, _ := grantsstate.EncodeGrant(grantsstate.Grant{}) + p.w.put(buckets.Grants, gkey, grant) + digest := "sha256:" + strings.Repeat("c", 64) + have, _ := nodeengineclient.EncodeHave(nodeengine.HaveByNodeModuleValue{Fp: fp, Composed: digest, Phase: nodeengineclient.PhaseRunning, Health: nodeengineclient.HealthOK}) + p.w.put(buckets.Have, "g14.web", have) + composes(t, p.w, "g14", digest) + placeOn(t, p.w, "g14", "web") + on, _ := presenceValue(true) + p.w.put(buckets.Presence, "g14", on) + if c.follow { + p.ends = newPipelineEnds(newMemEnds(), func() {}, t.Logf) + if c.passed { + _ = p.ends.take(pipe("web", trunkBranch, 5, 1, fp, mpr.PipelineEndedOutcomePassed), true) + } + } + p.wants(context.Background()) + key, _ := mwc.WantKey("g14", "web") + if _, got := wantsKV.values[key]; got != c.want { + t.Errorf("%s: a want written = %v, want %v", c.name, got, c.want) + } + if c.follow && c.passed { + raw, _, _ := p.ends.st.get(endKey("web", trunkBranch)) + var end End + _ = json.Unmarshal(raw, &end) + if !end.Written { + t.Errorf("%s: the end was not marked written", c.name) + } + } + } +} diff --git a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/tools.go b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/tools.go index d81e9acf..9b946428 100644 --- a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/tools.go +++ b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/tools.go @@ -404,6 +404,7 @@ func serveTools() error { tools = append(tools, viewTools(os.Getenv, keptSource, time.Now)...) // What each node's engine reads, as the assigner uses it (ADR 0618 §3): its own state engine-reads, asked again through it. tools = append(tools, enginereads.Tool(engineReadsTool, "the assigner", enginereads.KeptState(stdio.State(engineReadsState)), false, time.Now)) + tools = append(tools, writtenTools(runtimeStore{})...) tools = append(tools, askAgainTools(runtimeStore{}, stdio.Ask, time.Now)...) return stdio.Serve("", tools) } diff --git a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/world.go b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/world.go index 893d02f5..a5a1ca9a 100644 --- a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/world.go +++ b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/world.go @@ -578,3 +578,17 @@ func (w *world) unplacedNodes() []string { // trunkBranch is the branch of the trunk-only want, kept until the first registrations (design 47 G §95, §353). const trunkBranch = "main" + +// placedBranches is, by module and node, the watched branch each placement names ("" when it names none). +func (w *world) placedBranches() map[string]map[string]string { + out := map[string]map[string]string{} + for node, a := range w.assignments() { + for _, m := range a.Modules { + if out[m.Module] == nil { + out[m.Module] = map[string]string{} + } + out[m.Module][node] = m.Branch + } + } + return out +} diff --git a/modules/mesh-delivery-assigner/manifest_test.go b/modules/mesh-delivery-assigner/manifest_test.go index c1ceeaea..629dcee0 100644 --- a/modules/mesh-delivery-assigner/manifest_test.go +++ b/modules/mesh-delivery-assigner/manifest_test.go @@ -65,7 +65,7 @@ func TestManifest(t *testing.T) { } if want := map[string]int{"asks": 5, "differences": 64, "puts": 20, "wants": 20, "assignments": 64, "placement": 64, "copies": 1, "proposals": 5, "candidates": 5, "pulses": 0, "findings": 5, "policies": 20, "seat-holders": 5, views.MarksState: 1, views.StatusState: 1, - "engine-reads": 1}; !reflect.DeepEqual(states, want) { + "engine-reads": 1, "written": 5}; !reflect.DeepEqual(states, want) { t.Errorf("state = %v, want %v", states, want) } if _, ok := m.OwnSecrets["broker"]; !ok || len(m.OwnSecrets) != 1 { @@ -96,7 +96,7 @@ func TestManifest(t *testing.T) { // which lists and asks again what each node's engine reads. if !slices.Equal(m.Tools, []string{"assigner_put", "assigner_release", "assigner_puts", "assignments", "seat-holders", "wants", "assigner_read_assignments", "assigner_read_seat_holders", "assigner_read_wants", "assigner_read_findings", "assigner_read_asks", "assigner_read_differences", "assigner_read_copies", - "assigner_first_views", "assigner_first_view_again", "assigner_engine_reads", "assigner_seat_list", "assigner_seat_list_again", "assigner_ask_again"}) { + "assigner_first_views", "assigner_first_view_again", "assigner_engine_reads", "assigner_seat_list", "assigner_seat_list_again", "assigner_ask_again", "assigner_written"}) { t.Errorf("the assigner serves tools %v, want its three put tools, three first views, six read tools, assigner_engine_reads, the seat list's two (ADR 0636 §7) and assigner_ask_again (ADR 0668 §5)", m.Tools) } if m.RunsAs != m.Module || m.SecretsOwner != m.RunsAs { @@ -123,7 +123,7 @@ func TestManifest(t *testing.T) { } // ADR 0618 §4: and it hears each node's engine say its reads changed; ADR 0636 §4 (issue 1966): and the seat list's two events; // issue 1966 note 151: and the teller's ask-ended, its subject the one the assigner hears. - if !slices.Equal(m.Consumes, []string{"mesh-catalog.seat-definition-changed", "mesh-catalog.seat-definition-removed", "mesh-delivery-indexer.views.graph", "mesh-delivery-indexer.views.manifests", "mesh-delivery-indexer.views.modules", buckets.TellerModule + "." + tellerevents.AskEnded, "node-engine.manifest-reads-changed"}) { + if !slices.Equal(m.Consumes, []string{"mesh-catalog.seat-definition-changed", "mesh-catalog.seat-definition-removed", "mesh-delivery-indexer.views.graph", "mesh-delivery-indexer.views.manifests", "mesh-delivery-indexer.views.modules", buckets.TellerModule + "." + tellerevents.AskEnded, "mesh-pipeline-runner.pipeline-ended", "mesh-pipeline-runner.group-ended", "node-engine.manifest-reads-changed"}) { t.Errorf("consumes = %v, want the indexer's views of graph, manifests and modules, the teller's ask-ended and the engines' reads changed", m.Consumes) } if want := []string{"mesh-delivery-indexer.graph", "mesh-delivery-indexer.manifests", "mesh-delivery-indexer.modules", diff --git a/modules/mesh-delivery-assigner/module.json b/modules/mesh-delivery-assigner/module.json index e594b25c..d81cca86 100644 --- a/modules/mesh-delivery-assigner/module.json +++ b/modules/mesh-delivery-assigner/module.json @@ -45,6 +45,8 @@ "mesh-delivery-indexer.views.manifests", "mesh-delivery-indexer.views.modules", "mesh-delivery-teller.ask-ended", + "mesh-pipeline-runner.pipeline-ended", + "mesh-pipeline-runner.group-ended", "node-engine.manifest-reads-changed" ], "invokes": [ @@ -90,7 +92,8 @@ "assigner_engine_reads", "assigner_seat_list", "assigner_seat_list_again", - "assigner_ask_again" + "assigner_ask_again", + "assigner_written" ], "state": [ { "name": "assignments", "history": 64 }, @@ -108,7 +111,8 @@ { "name": "seat-holders", "history": 5 }, { "name": "view-marks", "history": 1 }, { "name": "first-views", "history": 1 }, - { "name": "engine-reads", "history": 1 } + { "name": "engine-reads", "history": 1 }, + { "name": "written", "history": 5 } ], "own-secrets": { "broker": "${dir:state}/broker" From 61277b6b9b47e651f8e7b19fcfbf2f9574d2ca4b Mon Sep 17 00:00:00 2001 From: jochen Date: Wed, 11 Nov 2026 11:53:41 +0100 Subject: [PATCH 08/10] VW4: the view's manifest says version 0.1, declares its finding stream and hears the keeper's ask MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Builds: design 47 G §383, §387, §368, §411 Claude-Session: https://claude.ai/code/session_01JG73KEq4zoDfM9WtYpcQgZ --- modules/mesh-delivery-view/manifest_test.go | 17 ++++++++++++++++- modules/mesh-delivery-view/module.json | 11 ++++++++++- 2 files changed, 26 insertions(+), 2 deletions(-) diff --git a/modules/mesh-delivery-view/manifest_test.go b/modules/mesh-delivery-view/manifest_test.go index a2e58671..b35cf6f8 100644 --- a/modules/mesh-delivery-view/manifest_test.go +++ b/modules/mesh-delivery-view/manifest_test.go @@ -51,6 +51,13 @@ type state struct { History int `json:"history,omitempty"` } +type stream struct { + Name string `json:"name"` + Events []string `json:"events"` + MaxAgeDays int `json:"max-age-days"` + MaxMiB int `json:"max-mib"` +} + type manifest struct { State []state `json:"state"` Module string `json:"module"` @@ -63,6 +70,7 @@ type manifest struct { Uses []string `json:"uses"` Tools []string `json:"tools"` Emits []string `json:"emits"` + Streams []stream `json:"streams"` OwnSecrets map[string]string `json:"own-secrets"` SecretsOwner string `json:"secrets-owner"` Resources []map[string]any `json:"resources"` @@ -117,7 +125,14 @@ func TestManifestIsTakenAndReadsOnly(t *testing.T) { // 1134/AC7 "seat:node-engine.manifest-reads"}; !reflect.DeepEqual(m.Invokes, want) { // ADR 0618 §3: node_manifest_reads asks the node-engine seat's verb of every node; 1585/T24: view_tiers asks indexer_pending, a verb edge (ADR 0614 §2) t.Errorf("invokes = %v, want %v", m.Invokes, want) } - if want := []string{"mesh-delivery-assigner.views.asks", "mesh-delivery-assigner.views.assignments", "mesh-delivery-assigner.views.wants", + // design 47 G §387: the keeper's ask for the findings is heard, the finding stream declared (§383), and §368: version is MAJOR.MINOR. + if want := []stream{{Name: "mesh-delivery-view-findings", Events: []string{"finding-written.*", "finding-cleared.*", "findings-stated.*"}, MaxAgeDays: 7, MaxMiB: 16}}; !reflect.DeepEqual(m.Streams, want) { + t.Errorf("streams = %+v, want %+v", m.Streams, want) + } + if m.Version != "0.1" { + t.Errorf("version = %q, want 0.1", m.Version) + } + if want := []string{"mesh-conditions.findings-asked", "mesh-delivery-assigner.views.asks", "mesh-delivery-assigner.views.assignments", "mesh-delivery-assigner.views.wants", "mesh-delivery-indexer.views.candidates", "mesh-delivery-indexer.views.modules", "mesh-delivery-indexer.views.taken"}; !reflect.DeepEqual(m.Consumes, want) { t.Errorf("consumes = %v, want %v", m.Consumes, want) } diff --git a/modules/mesh-delivery-view/module.json b/modules/mesh-delivery-view/module.json index 81d90b74..a8544e42 100644 --- a/modules/mesh-delivery-view/module.json +++ b/modules/mesh-delivery-view/module.json @@ -1,6 +1,6 @@ { "module": "mesh-delivery-view", - "version": "1", + "version": "0.1", "moved": true, "slug": "view", "tools": [ @@ -42,6 +42,7 @@ { "name": "findings", "history": 1 } ], "consumes": [ + "mesh-conditions.findings-asked", "mesh-delivery-assigner.views.asks", "mesh-delivery-assigner.views.assignments", "mesh-delivery-assigner.views.wants", @@ -49,6 +50,14 @@ "mesh-delivery-indexer.views.modules", "mesh-delivery-indexer.views.taken" ], + "streams": [ + { + "name": "mesh-delivery-view-findings", + "events": ["finding-written.*", "finding-cleared.*", "findings-stated.*"], + "max-age-days": 7, + "max-mib": 16 + } + ], "invokes": [ "mesh-delivery-assigner.assigner_read_asks", "mesh-delivery-assigner.assigner_read_differences", From 1778888acca14459a3ccc23ecceda029921cd578 Mon Sep 17 00:00:00 2001 From: jochen Date: Wed, 11 Nov 2026 11:54:50 +0100 Subject: [PATCH 09/10] AS2: the wants keep the trunk-only form until the first registration, and follow the pipelines' ends from it MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Builds: design 47 G §95, §130, §353, §352 Claude-Session: https://claude.ai/code/session_01JG73KEq4zoDfM9WtYpcQgZ --- .../cmd/mesh-delivery-assigner/main.go | 10 +- .../cmd/mesh-delivery-assigner/pass.go | 4 +- .../cmd/mesh-delivery-assigner/pipelines.go | 96 ++++++++++++++++++- .../mesh-delivery-assigner/pipelines_test.go | 39 ++++++++ .../mesh-delivery-assigner/manifest_test.go | 4 +- modules/mesh-delivery-assigner/module.json | 2 + 6 files changed, 143 insertions(+), 12 deletions(-) diff --git a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/main.go b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/main.go index 4f040bed..3ed9dab2 100644 --- a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/main.go +++ b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/main.go @@ -493,9 +493,13 @@ func follow(ctx context.Context, js jetstream.JetStream, nc process.Subscriber, if kv := kvs[writtenBucket]; kv != nil { // The runner's pipeline-ended and group-ended (design 47 G §29): kept in the state written, which the wants follow // once the first registrations are written (§95, §353). - p.heard = newPipelineEnds(kvEnds{ctx, kv}, wake, logger.Printf) - if err := hearEnds(hub, p.heard); err != nil { - logger.Printf("a pipeline's end is not heard, so no want follows a pipeline until the assigner starts again: %v", err) + p.ends = newPipelineEnds(kvEnds{ctx, kv}, wake, logger.Printf) + err := hearEnds(hub, p.ends) + if err == nil && bus != nil { + err = followRegistrations(ctx, hub, bus.Call, p.ends) + } + if err != nil { + logger.Printf("a pipeline's end or a registration is not heard, so the wants keep the trunk-only form until the assigner starts again: %v", err) } } hearReadsChanged(ctx, js, hub, reads, findings, gate.eventsLive, logger.Printf) diff --git a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/pass.go b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/pass.go index 69d369af..8c5d6aa2 100644 --- a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/pass.go +++ b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/pass.go @@ -58,10 +58,8 @@ type passer struct { // neither is, by node and module, each module on neither list whose condition this process raised (issue 1945). Nil until the // first pass after a start, which clears what a previous process may have raised (startClears). neither map[string]neitherOne - // ends are the pipeline ends the wants follow (pipelines.go); nil is the trunk-only want, kept until the first registrations. + // ends are the pipeline ends the wants follow (pipelines.go), once the first registrations are heard; until then, and nil, the trunk-only want. ends *pipelineEnds - // heard is the same follower while the wants do not follow it yet (ends is nil). - heard *pipelineEnds // acts are the puts' acts (ADR 0546); nil: the puts are not taken further (no bus to call, or no puts bucket). acts *putActs // noWant is, by node and module, the reason last said why the node gets no want for the module (issue 1945): a reason is diff --git a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/pipelines.go b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/pipelines.go index 743125a5..5e9ab67e 100644 --- a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/pipelines.go +++ b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/pipelines.go @@ -5,13 +5,19 @@ import ( "encoding/json" "errors" "fmt" + "slices" "sort" + "strings" "sync" + "sync/atomic" + "time" "github.com/nats-io/nats.go/jetstream" + mcc "git.novox.be/novox/mesh-delivery/contracts/go/clients/meshcatalogclient" mprc "git.novox.be/novox/mesh-delivery/contracts/go/clients/meshpipelinerunnerclient" mwc "git.novox.be/novox/mesh-delivery/contracts/go/clients/meshwantsclient" + mpcat "git.novox.be/novox/mesh-delivery/contracts/go/seats/meshcatalog" mpr "git.novox.be/novox/mesh-delivery/contracts/go/seats/meshpipelinerunner" stdio "git.novox.be/novox/mesh-sdk/go" "git.novox.be/novox/mesh-sdk/go/keys" @@ -81,6 +87,9 @@ type endStore interface { get(key string) (value []byte, revision uint64, err error) // swap writes value at key if the key stands at revision (0: if it does not stand); errSwapped when it does not. swap(key string, value []byte, revision uint64) error + // keys are the keys that stand, and drop deletes one. + keys() ([]string, error) + drop(key string) error } var errSwapped = errors.New("the key changed") @@ -115,6 +124,16 @@ func (k kvEnds) swap(key string, value []byte, revision uint64) error { return err } +func (k kvEnds) keys() ([]string, error) { + ks, err := k.kv.Keys(k.ctx) + if errors.Is(err, jetstream.ErrNoKeysFound) { + return nil, nil + } + return ks, err +} + +func (k kvEnds) drop(key string) error { return k.kv.Delete(k.ctx, key) } + func isWrongSequence(err error) bool { var ae *jetstream.APIError return errors.As(err, &ae) && ae.ErrorCode == jetstream.JSErrCodeStreamWrongLastSequence @@ -126,7 +145,9 @@ type pipelineEnds struct { st endStore wake func() logf func(string, ...any) - mu sync.Mutex + // on is whether the wants follow the ends: set by the first registration heard, never before (design 47 G §95, §353). + on atomic.Bool + mu sync.Mutex // firsts is, by module, the first watched branch of its registration, the branch of a placement that names none (design // 47 G §14). firsts map[string]string @@ -278,7 +299,7 @@ func writtenTools(st store) []stdio.Tool { // branchesOf is the watched branches the placements of module chose, each once; the default branch when none names one. func (e *pipelineEnds) branchesOf(module string, placed map[string]map[string]string) []string { - if e == nil { + if !e.following() { return nil } seen := map[string]bool{} @@ -300,7 +321,7 @@ func (e *pipelineEnds) branchesOf(module string, placed map[string]map[string]st // mayWant says whether module's wants may name fp: a pipeline passed at fp on every branch, or the wants name fp already // (a node placed later, a module taken away). func (e *pipelineEnds) mayWant(module, fp string, branches []string, wanted map[string]mwc.Want) bool { - if e == nil { + if !e.following() { return true } for _, w := range wanted { @@ -318,10 +339,77 @@ func (e *pipelineEnds) mayWant(module, fp string, branches []string, wanted map[ // wrote marks the end the wants of module were written from. func (e *pipelineEnds) wrote(module, fp string, branches []string) { - if e == nil { + if !e.following() { return } for _, b := range branches { e.markWritten(module, b, fp) } } + +// following says whether the wants follow the pipeline ends yet; a nil follower never does. +func (e *pipelineEnds) following() bool { return e != nil && e.on.Load() } + +// registered takes a module's registration (design 47 G §353): the first one heard switches the wants to the ends, its first +// watched branch is the default a placement runs, and the ends kept for a branch it no longer watches are deleted (§352). gone +// is a registration removed: all its ends go. +func (e *pipelineEnds) registered(reg mpcat.Registration, gone bool) { + e.mu.Lock() + if gone || len(reg.Branches) == 0 { + delete(e.firsts, reg.Module) + } else { + e.firsts[reg.Module] = reg.Branches[0] + } + e.mu.Unlock() + kept, err := e.st.keys() + if err != nil { + e.logf("the ends kept cannot be listed: %v", err) + } + prefix := keys.EscapeName(reg.Module) + "." + for _, k := range kept { + rest, ok := strings.CutPrefix(k, prefix) + if !ok { + continue + } + branch, err := keys.UnescapeName(rest) + if err == nil && !gone && slices.Contains(reg.Branches, branch) { + continue + } + if err := e.st.drop(k); err != nil { + e.logf("the end %s is not deleted: %v", k, err) + } + } + if !e.on.Swap(true) { + e.logf("the first registration is heard (%s): the wants follow the pipelines' ends from now", reg.Module) + } + e.wake() +} + +// followRegistrations follows the registrations through the seat mesh-catalog (design 47 G §353): a first view of its verb +// and each registration-changed on the hub. It returns when the hub's handler is registered; the first view is asked on its own +// goroutine, the wants keep the trunk-only form until a registration is heard. +func followRegistrations(ctx context.Context, hub *process.EventHub, call mcc.Call, e *pipelineEnds) error { + regs := mcc.NewRegistrations(call) + regs.Follower.Logf = e.logf + regs.Follower.Changed = func(module string, value json.RawMessage, gone bool, _ time.Time) { + reg := mpcat.Registration{Module: module} + if !gone { + _ = json.Unmarshal(value, ®) + } + e.registered(reg, gone) + } + if err := hub.Handle(mpcat.SubjectRegistrationChanged, func(m jetstream.Msg) { + if err := regs.Hear(ctx, m.Data()); err != nil { + e.logf("%s: %v", m.Subject(), err) + } + _ = m.Ack() + }); err != nil { + return err + } + go func() { + if err := regs.Start(ctx); err != nil && ctx.Err() == nil { + e.logf("the first view of the registrations is not answered: %v", err) + } + }() + return nil +} diff --git a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/pipelines_test.go b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/pipelines_test.go index 17597d49..acb0c207 100644 --- a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/pipelines_test.go +++ b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/pipelines_test.go @@ -10,6 +10,7 @@ import ( "git.novox.be/novox/mesh-delivery/contracts/go/clients/meshcatalogclient" mwc "git.novox.be/novox/mesh-delivery/contracts/go/clients/meshwantsclient" "git.novox.be/novox/mesh-delivery/contracts/go/seats/meshcatalog" + mpcat "git.novox.be/novox/mesh-delivery/contracts/go/seats/meshcatalog" mpr "git.novox.be/novox/mesh-delivery/contracts/go/seats/meshpipelinerunner" oc "git.novox.be/novox/mesh-contracts/operator-and-conversation/go/seats/operatorchannel" @@ -34,6 +35,13 @@ type memEnds struct { func newMemEnds() *memEnds { return &memEnds{vals: map[string][]byte{}, revs: map[string]uint64{}} } func (m *memEnds) get(key string) ([]byte, uint64, error) { return m.vals[key], m.revs[key], nil } +func (m *memEnds) keys() (out []string, _ error) { + for k := range m.vals { + out = append(out, k) + } + return out, nil +} +func (m *memEnds) drop(key string) error { delete(m.vals, key); delete(m.revs, key); return nil } func (m *memEnds) swap(key string, v []byte, rev uint64) error { if m.revs[key] != rev { return errSwapped @@ -79,6 +87,7 @@ func TestAWantFollowsAPassedPipeline(t *testing.T) { } { woken := 0 e := newPipelineEnds(newMemEnds(), func() { woken++ }, t.Logf) + e.on.Store(true) for _, p := range c.ends { if p.Group != "" { continue // heard and ignored, as hearPipeline does @@ -100,12 +109,14 @@ func TestAWantFollowsAPassedPipeline(t *testing.T) { } // A want that names the fingerprint already is not held by a later failure (a node placed afterwards gets it). e := newPipelineEnds(newMemEnds(), func() {}, t.Logf) + e.on.Store(true) if !e.mayWant("web", fp1, []string{"main"}, map[string]mwc.Want{"g14": {Fp: fp1}}) { t.Error("a want naming the fingerprint was held") } // markWritten keeps the end and says it was written from; a newer end replaces it unwritten. st := newMemEnds() e = newPipelineEnds(st, func() {}, t.Logf) + e.on.Store(true) _ = e.take(pipe("web", "main", 5, 1, fp1, pass), true) e.wrote("web", fp1, []string{"main"}) var end End @@ -153,6 +164,7 @@ func TestAPassWritesNoWantWithoutAPassedPipeline(t *testing.T) { p.w.put(buckets.Presence, "g14", on) if c.follow { p.ends = newPipelineEnds(newMemEnds(), func() {}, t.Logf) + p.ends.on.Store(true) if c.passed { _ = p.ends.take(pipe("web", trunkBranch, 5, 1, fp, mpr.PipelineEndedOutcomePassed), true) } @@ -172,3 +184,30 @@ func TestAPassWritesNoWantWithoutAPassedPipeline(t *testing.T) { } } } + +// Design 47 G §95, §130, §353, §352 (test 100): the wants keep the trunk-only form until the first registration is heard and +// switch on it; a branch a registration no longer watches loses its end, and its first branch is a placement's default. +func TestTheWantsSwitchOnTheFirstRegistration(t *testing.T) { + st := newMemEnds() + woken := 0 + e := newPipelineEnds(st, func() { woken++ }, t.Logf) + _ = e.take(pipe("web", "main", 5, 1, "fp1", mpr.PipelineEndedOutcomeFailed), false) + _ = e.take(pipe("web", "beta", 5, 1, "fp1", mpr.PipelineEndedOutcomePassed), true) + if e.following() || !e.mayWant("web", "fp1", []string{"main"}, nil) { + t.Fatal("the follower held a want before any registration") + } + e.registered(mpcat.Registration{Module: "web", Branches: []string{"beta"}}, false) + if !e.following() || woken == 0 { + t.Error("the first registration did not switch the wants on") + } + if e.mayWant("web", "fp1", []string{"main"}, nil) || !e.mayWant("web", "fp1", []string{"beta"}, nil) { + t.Error("after the switch the wants did not follow the ends") + } + if len(st.vals) != 1 || e.branchOf("web", "") != "beta" || e.branchOf("other", "") != trunkBranch { + t.Errorf("ends kept %d, default branch %q: an unwatched branch's end stayed, or the default is wrong", len(st.vals), e.branchOf("web", "")) + } + e.registered(mpcat.Registration{Module: "web"}, true) + if len(st.vals) != 0 { + t.Errorf("a registration gone left %d ends", len(st.vals)) + } +} diff --git a/modules/mesh-delivery-assigner/manifest_test.go b/modules/mesh-delivery-assigner/manifest_test.go index 629dcee0..036dbc34 100644 --- a/modules/mesh-delivery-assigner/manifest_test.go +++ b/modules/mesh-delivery-assigner/manifest_test.go @@ -123,11 +123,11 @@ func TestManifest(t *testing.T) { } // ADR 0618 §4: and it hears each node's engine say its reads changed; ADR 0636 §4 (issue 1966): and the seat list's two events; // issue 1966 note 151: and the teller's ask-ended, its subject the one the assigner hears. - if !slices.Equal(m.Consumes, []string{"mesh-catalog.seat-definition-changed", "mesh-catalog.seat-definition-removed", "mesh-delivery-indexer.views.graph", "mesh-delivery-indexer.views.manifests", "mesh-delivery-indexer.views.modules", buckets.TellerModule + "." + tellerevents.AskEnded, "mesh-pipeline-runner.pipeline-ended", "mesh-pipeline-runner.group-ended", "node-engine.manifest-reads-changed"}) { + if !slices.Equal(m.Consumes, []string{"mesh-catalog.registration-changed", "mesh-catalog.seat-definition-changed", "mesh-catalog.seat-definition-removed", "mesh-delivery-indexer.views.graph", "mesh-delivery-indexer.views.manifests", "mesh-delivery-indexer.views.modules", buckets.TellerModule + "." + tellerevents.AskEnded, "mesh-pipeline-runner.pipeline-ended", "mesh-pipeline-runner.group-ended", "node-engine.manifest-reads-changed"}) { t.Errorf("consumes = %v, want the indexer's views of graph, manifests and modules, the teller's ask-ended and the engines' reads changed", m.Consumes) } if want := []string{"mesh-delivery-indexer.graph", "mesh-delivery-indexer.manifests", "mesh-delivery-indexer.modules", - "seat:mesh-catalog.seat-definitions", // ADR 0636 §4: the seat list + "seat:mesh-catalog.registrations", "seat:mesh-catalog.seat-definitions", // ADR 0636 §4: the seat list "seat:mesh-hand-acts.hand-act", "seat:mesh-hand-acts.hand-acts", "seat:mesh-hand-acts.warranted", // ADR 0565 §9, §11: the seat alone "seat:node-backup.backed-up", "seat:node-backup.now", "seat:node-engine.manifest-reads", // ADR 0618 §3 diff --git a/modules/mesh-delivery-assigner/module.json b/modules/mesh-delivery-assigner/module.json index d81cca86..d6252329 100644 --- a/modules/mesh-delivery-assigner/module.json +++ b/modules/mesh-delivery-assigner/module.json @@ -39,6 +39,7 @@ "operator-channel" ], "consumes": [ + "mesh-catalog.registration-changed", "mesh-catalog.seat-definition-changed", "mesh-catalog.seat-definition-removed", "mesh-delivery-indexer.views.graph", @@ -53,6 +54,7 @@ "mesh-delivery-indexer.graph", "mesh-delivery-indexer.manifests", "mesh-delivery-indexer.modules", + "seat:mesh-catalog.registrations", "seat:mesh-catalog.seat-definitions", "seat:mesh-hand-acts.hand-act", "seat:mesh-hand-acts.hand-acts", From 91a9c4129056bcd4bdd038074f72eccd5b657ecd Mon Sep 17 00:00:00 2001 From: jochen Date: Wed, 11 Nov 2026 11:55:02 +0100 Subject: [PATCH 10/10] gofmt --- modules/mesh-delivery-assigner/manifest_test.go | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/modules/mesh-delivery-assigner/manifest_test.go b/modules/mesh-delivery-assigner/manifest_test.go index 036dbc34..1db630c4 100644 --- a/modules/mesh-delivery-assigner/manifest_test.go +++ b/modules/mesh-delivery-assigner/manifest_test.go @@ -127,7 +127,7 @@ func TestManifest(t *testing.T) { t.Errorf("consumes = %v, want the indexer's views of graph, manifests and modules, the teller's ask-ended and the engines' reads changed", m.Consumes) } if want := []string{"mesh-delivery-indexer.graph", "mesh-delivery-indexer.manifests", "mesh-delivery-indexer.modules", - "seat:mesh-catalog.registrations", "seat:mesh-catalog.seat-definitions", // ADR 0636 §4: the seat list + "seat:mesh-catalog.registrations", "seat:mesh-catalog.seat-definitions", // ADR 0636 §4: the seat list "seat:mesh-hand-acts.hand-act", "seat:mesh-hand-acts.hand-acts", "seat:mesh-hand-acts.warranted", // ADR 0565 §9, §11: the seat alone "seat:node-backup.backed-up", "seat:node-backup.now", "seat:node-engine.manifest-reads", // ADR 0618 §3