From d5676ee4b3ca9e425a816cdcbeda55512abe04e6 Mon Sep 17 00:00:00 2001 From: jochen Date: Thu, 12 Nov 2026 09:14:45 +0100 Subject: [PATCH] AS12: the assigner keeps a run's candidate assignment graph, dropped on the run's group-ended MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A run's candidate catalog-revision (meshcatalogclient's candidates hook) is read whole through the catalog seat's revision verb; the assignment graph is computed from it in place of its branch's half and kept apart from trunk's under candidate.., the group run-. revision, entry and difference serve it (a candidate against trunk). The runner's group-ended of the run drops it and writes no want; the gate's heads-replaced drops nothing; its entries stay while it is kept. Builds: design 47 G §189, §190, §186, §187, §359, A-201, A-235, A-160 Claude-Session: https://claude.ai/code/session_01NcbgfGKaxw984XyAPfLm4h --- .../cmd/mesh-delivery-assigner/candidates.go | 113 ++++++++++++++ .../cmd/mesh-delivery-assigner/graphkept.go | 22 ++- .../mesh-delivery-assigner/graphkept_test.go | 44 ++++++ .../cmd/mesh-delivery-assigner/main.go | 6 + .../cmd/mesh-delivery-assigner/pins.go | 14 ++ .../cmd/mesh-delivery-assigner/pins_test.go | 12 ++ .../cmd/mesh-delivery-assigner/pipelines.go | 16 +- .../mesh-delivery-assigner/pipelines_test.go | 43 ++++++ .../cmd/mesh-delivery-assigner/views.go | 16 ++ .../internal/assigngraph/candidate.go | 138 ++++++++++++++++++ .../internal/assigngraph/graph_test.go | 74 ++++++++++ .../internal/assigngraph/keep.go | 58 ++++---- .../internal/assigngraph/verbs.go | 26 +++- 13 files changed, 543 insertions(+), 39 deletions(-) create mode 100644 modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/candidates.go create mode 100644 modules/mesh-delivery-assigner/internal/assigngraph/candidate.go diff --git a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/candidates.go b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/candidates.go new file mode 100644 index 00000000..c19cfb45 --- /dev/null +++ b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/candidates.go @@ -0,0 +1,113 @@ +package main + +import ( + "context" + "encoding/json" + "errors" + "fmt" + "maps" + "strings" + + mcc "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/clients/meshcatalogclient" + "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/seats/meshcatalog" + "git.novox.be/novox/mesh-sdk/go/buscalls" + + "mesh-delivery-assigner/internal/assigngraph" +) + +// A run's candidate assignment graph (design 47 G §189, §190, A-201, A-235): the candidate `run-`'s catalog graph revision, +// handed by meshcatalogclient's candidates hook, is read whole through the catalog seat's verb revision; the assignment graph is +// computed from it in place of its branch's half, beside every other followed branch's, and kept under candidate..; +// the runner's group-ended of the run drops it. It writes no want and says no event; the gate's heads-replaced drops nothing. + +// runGroupPrefix is a run on request's candidate group before its id, `run-` (A-160). +const runGroupPrefix = "run-" + +// readCandidateHalf reads a candidate's catalog graph revision whole through the seat's verb revision, page by page, with the +// candidate's group and heads and the revision its event names. +func readCandidateHalf(call mcc.Call, ev meshcatalog.CatalogRevisionEvent) (mcc.Half, error) { + half := mcc.Half{Entries: map[string]meshcatalog.CatalogEntry{}} + after := "" + for { + a, err := askCandidatePage(call, meshcatalog.RevisionInput{Branch: ev.Branch, Group: ev.Group, Heads: ev.Heads, Revision: ev.Revision, After: after}) + if err != nil { + return half, err + } + if a.Revision == nil || a.Revision.Revision != ev.Revision || a.Revision.Group != ev.Group { + return half, fmt.Errorf("%s answered %+v for the candidate %s's revision %d", mcc.RevisionVerb, a.Revision, ev.Group, ev.Revision) + } + half.Revision = *a.Revision + for _, e := range a.Entries { + half.Entries[e.Module] = e + } + if a.Next == "" { + return half, nil + } + after = a.Next + } +} + +// keepCandidate computes and keeps the candidate's assignment graph of a run's catalog-revision; another candidate's (a try's, a +// group's) is passed over. +func (p *passer) keepCandidate(ctx context.Context, call mcc.Call, ev meshcatalog.CatalogRevisionEvent) error { + if !strings.HasPrefix(ev.Group, runGroupPrefix) { + return nil + } + g := p.graph + if g == nil || g.objects == nil || g.halves == nil { + return errors.New("the assigner keeps no assignment graph, so no candidate's is kept") + } + half, err := readCandidateHalf(call, ev) + if err != nil { + return fmt.Errorf("the candidate %s's revision %d is not read, so its assignment graph is not kept: %w", ev.Group, ev.Revision, err) + } + in := p.graphInputs() + in.Halves = maps.Clone(in.Halves) + in.Halves[half.Revision.Branch] = half + c := assigngraph.Compute(in) + for _, u := range c.Uncomputed { + p.logf("the candidate %s's assignment graph leaves out %s", ev.Group, u) + } + rev, fresh, err := assigngraph.KeepCandidate(ctx, g.objects, c, ev.Group, ev.Heads, meshcatalog.EventCatalogRevision, p.clock()) + if err != nil { + return fmt.Errorf("the candidate %s's assignment graph is not kept: %w", ev.Group, err) + } + if fresh { + p.logf("the candidate %s's assignment graph revision %d is kept, from its catalog graph revision %d", ev.Group, rev.Revision, ev.Revision) + } + return nil +} + +// dropCandidate removes a run's candidate assignment graph on the runner's group-ended of that run (A-235). +func (p *passer) dropCandidate(ctx context.Context, group string) error { + g := p.graph + if g == nil || g.objects == nil { + return nil + } + o, ok := g.objects.(assigngraph.Remover) + if !ok { + return fmt.Errorf("the objects %s remove nothing, so the candidate %s is kept", assigngraph.RevisionsObjects, group) + } + n, err := assigngraph.DropCandidate(ctx, o, group) + if err == nil && n > 0 { + p.logf("the candidate %s's assignment graph is dropped: its run ended", group) + } + return err +} + +// askCandidatePage is one page of a candidate's revision as the seat's verb revision answers it. +func askCandidatePage(call mcc.Call, in meshcatalog.RevisionInput) (meshcatalog.RevisionAnswer, error) { + raw, err := call(mcc.RevisionVerb, in) + if err != nil { + return meshcatalog.RevisionAnswer{}, err + } + answer, said, ok := buscalls.Unwrap(raw) + if !ok { + return meshcatalog.RevisionAnswer{}, fmt.Errorf("%s answered: %s", mcc.RevisionVerb, said) + } + var a meshcatalog.RevisionAnswer + if err := json.Unmarshal(answer, &a); err != nil { + return a, fmt.Errorf("%s answered no revision: %w", mcc.RevisionVerb, err) + } + return a, nil +} diff --git a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/graphkept.go b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/graphkept.go index 50e38343..7a6c49c7 100644 --- a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/graphkept.go +++ b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/graphkept.go @@ -44,14 +44,7 @@ func (p *passer) keepGraph(ctx context.Context) { if g == nil || g.objects == nil || g.halves == nil { return } - in := assigngraph.Inputs{Halves: g.halves(), Placements: p.w.assignments(), Capabilities: p.capabilityDigests()} - // With no pipeline ends kept, the wants are the trunk-only form (pass.go): a placement naming no branch runs the trunk's. - first := p.w.trunkName() - in.FirstBranch = func(string) string { return first } - if p.ends != nil { - in.FirstBranch = func(module string) string { return p.ends.branchOf(module, "") } - } - c := assigngraph.Compute(in) + c := assigngraph.Compute(p.graphInputs()) for _, u := range c.Uncomputed { if g.said != u { g.said = u @@ -85,6 +78,19 @@ func (p *passer) keepGraph(ctx context.Context) { } } +// graphInputs are what the pass's assignment graph is computed from: every followed branch's catalog half, the placements, each +// node's capability reading and the first branch of each registration. +func (p *passer) graphInputs() assigngraph.Inputs { + in := assigngraph.Inputs{Halves: p.graph.halves(), Placements: p.w.assignments(), Capabilities: p.capabilityDigests()} + // With no pipeline ends kept, the wants are the trunk-only form (pass.go): a placement naming no branch runs the trunk's. + first := p.w.trunkName() + in.FirstBranch = func(string) string { return first } + if p.ends != nil { + in.FirstBranch = func(module string) string { return p.ends.branchOf(module, "") } + } + return in +} + // capabilityDigests is the digest of each node's capability reading, the statement its engine made, by node. func (p *passer) capabilityDigests() []meshwants.AssignmentRevisionCapabilitiesItem { if p.reads == nil { diff --git a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/graphkept_test.go b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/graphkept_test.go index ad7637be..973a66ec 100644 --- a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/graphkept_test.go +++ b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/graphkept_test.go @@ -12,6 +12,7 @@ import ( "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/seats/meshcatalog" "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/seats/meshwants" + "mesh-delivery-assigner/internal/assigngraph" "mesh-delivery-assigner/internal/buckets" ) @@ -77,4 +78,47 @@ func TestTheWantsAreSaidWantedFromTheKeptAssignmentGraphRevision(t *testing.T) { if err := json.Unmarshal(events[len(events)-1].body, &w); err != nil || len(events) != 3 || !w.Removed || w.Fp != fpTest { t.Errorf("a removing want said %+v (%v), %d events", w, err, len(events)) } + + // A-201, A-235: a run's candidate catalog revision, read whole through the seat's verb revision, keeps the run's assignment + // graph under candidate.. in place of its branch's half; it says nothing and writes no want, a try's candidate + // keeps none, and the run's group-ended drops it. + before, puts := len(events), kv.puts + ev := meshcatalog.CatalogRevisionEvent{Kind: "candidate", Branch: trunkBranch, Group: "run-7", Heads: "h1", Revision: 5} + var asked []meshcatalog.RevisionInput + call := func(key string, body any) (json.RawMessage, error) { + in := body.(meshcatalog.RevisionInput) + asked = append(asked, in) + rev := meshcatalog.CatalogRevision{Branch: trunkBranch, Kind: "candidate", Group: "run-7", Heads: "h1", Revision: 5, Source: 9, + Entries: []meshcatalog.EntryRef{{Module: "web", Digest: "sha256:" + strings.Repeat("b", 64)}}} + a := meshcatalog.RevisionAnswer{Revision: &rev, Entries: []meshcatalog.CatalogEntry{{Module: "web", Fp: "src2:w2"}}} + return json.Marshal(a) + } + if err := p.keepCandidate(ctx, call, meshcatalog.CatalogRevisionEvent{Kind: "candidate", Branch: trunkBranch, Group: "try-3", Heads: "h", Revision: 2}); err != nil || len(asked) != 0 { + t.Errorf("a try's candidate asked %v: %v", asked, err) + } + if err := p.keepCandidate(ctx, call, ev); err != nil { + t.Fatal(err) + } + if len(asked) != 1 || asked[0].Group != "run-7" || asked[0].Heads != "h1" || asked[0].Revision != 5 { + t.Errorf("the candidate's revision was asked as %+v", asked) + } + a, err := assigngraph.Revision(ctx, objects, meshwants.RevisionInput{Group: "run-7"}) + if err != nil || a.Revision.Kind != "candidate" || len(a.Revision.Catalogs) != 1 || a.Revision.Catalogs[0].Revision != 5 { + t.Errorf("the run's assignment graph is %+v: %v", a.Revision, err) + } + if e, err := assigngraph.Entry(ctx, objects, meshwants.EntryInput{Group: "run-7", Node: "ace", Module: "web"}); err != nil || e.Entry.Fp != "src2:w2" { + t.Errorf("the run's entry of web is %+v: %v", e.Entry, err) + } + if len(events) != before || kv.puts != puts { + t.Errorf("a candidate said %d events and wrote %d wants", len(events)-before, kv.puts-puts) + } + if err := p.dropCandidate(ctx, "run-7"); err != nil { + t.Fatal(err) + } + if _, ok := objects[assigngraph.CandidateObject("run-7", "h1")]; ok { + t.Error("the run's group-ended left its candidate kept") + } + if _, err := assigngraph.Revision(ctx, objects, meshwants.RevisionInput{}); err != nil { + t.Errorf("dropping the run's candidate dropped trunk: %v", err) + } } 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 976b5bfe..982be1ce 100644 --- a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/main.go +++ b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/main.go @@ -29,6 +29,7 @@ import ( mcseat "git.novox.be/novox/mesh-contracts/health-and-repair/go/seats/meshconditions" occ "git.novox.be/novox/mesh-contracts/operator-and-conversation/go/clients/operatorchannelclient" oc "git.novox.be/novox/mesh-contracts/operator-and-conversation/go/seats/operatorchannel" + "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/seats/meshcatalog" "git.novox.be/novox/mesh-sdk/go/announce" "git.novox.be/novox/mesh-sdk/go/process" "git.novox.be/novox/mesh-sdk/go/seatlist" @@ -559,6 +560,10 @@ func follow(ctx context.Context, js jetstream.JetStream, nc process.Subscriber, if iv != nil && iv.h != nil { objects := newBusObjects(js, process.Account{Node: thisNode, Module: assignment.Module}, assigngraph.RevisionsObjects) p.graph = &graphKeeper{objects: objects, halves: iv.h.views} + keep := func(ctx context.Context, ev meshcatalog.CatalogRevisionEvent) error { + return p.keepCandidate(ctx, iv.call, ev) + } + iv.candidates.Store(&keep) // The pins of the revisions (pins.go, design 47 G §579): the gate's pinned, freed, heads-replaced and pins-stated // on the hub, registered before it starts. if kv := kvs[pinsBucket]; kv != nil { @@ -590,6 +595,7 @@ func follow(ctx context.Context, js jetstream.JetStream, nc process.Subscriber, // once the first registrations are written (§95, §353). p.ends = newPipelineEnds(kvEnds{ctx, kv}, wake, logger.Printf) p.ends.trunk = w.trunkName + p.ends.runEnded = func(group string) error { return p.dropCandidate(ctx, group) } err := hearEnds(hub, p.ends) if err == nil && bus != nil { err = followRegistrations(ctx, hub, bus.Call, p.ends) diff --git a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/pins.go b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/pins.go index 0ccafce5..fa6055ac 100644 --- a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/pins.go +++ b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/pins.go @@ -365,6 +365,20 @@ func (k *pinKeeper) remove() error { if removed == 0 { return nil } + // A run's candidate kept now lists its entries too, until its group-ended drops it (A-201, A-235). + for _, n := range names { + group, heads, ok := meshwants.ParseAssignmentRevisionsByCandidateGroupHeadsKey(n) + if !ok { + continue + } + rev, err := assigngraph.ReadCandidate(k.ctx, k.objects, group, heads) + if err != nil { + return err + } + for _, e := range rev.Entries { + listed[e.Digest] = true + } + } for _, n := range names { digest, ok := strings.CutPrefix(n, "entry.") if !ok || listed[digest] { diff --git a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/pins_test.go b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/pins_test.go index de19ca35..de2a17d8 100644 --- a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/pins_test.go +++ b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/pins_test.go @@ -38,6 +38,11 @@ func TestPinsKeepARevisionWhileARecordPinsIt(t *testing.T) { raw, _ := json.Marshal(rev) objects[assigngraph.TrunkObject(n)] = raw } + // A run's candidate lists an entry no trunk revision lists (A-201): it stays while the candidate is kept. + cand := entry("d", "fp-d2") + candRaw, _ := json.Marshal(meshwants.AssignmentRevision{Revision: 1, Kind: "candidate", Group: "run-7", Heads: "h1", + Entries: []meshwants.NodeEntryRef{{Node: "n1", Module: "d", Digest: cand}}}) + objects[assigngraph.CandidateObject("run-7", "h1")] = candRaw revision(1, shared, old, running) revision(2, shared, old) revision(3, shared) @@ -99,6 +104,13 @@ func TestPinsKeepARevisionWhileARecordPinsIt(t *testing.T) { if _, ok := objects[assigngraph.EntryObject(shared)]; !ok { t.Error("the entry the current revision lists is removed") } + if _, ok := objects[assigngraph.EntryObject(cand)]; !ok { + t.Error("the entry a run's kept candidate lists is removed") + } + say(mdseat.SubjectHeadsReplaced, mdseat.HeadsReplacedEvent{}) + if _, ok := objects[assigngraph.CandidateObject("run-7", "h1")]; !ok { + t.Error("the gate's heads-replaced dropped a run's candidate") + } say(mdseat.SubjectPinned, pins("r3", 2)) raw, _, _ = k.st.get(pinKey(2)) if string(raw) != `{"records":["r3"],"missing":true}` || len(found) != 1 || found[0].Kind != pinMissingKind || found[0].ID != "2" { 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 8ecc9bec..cedaedfa 100644 --- a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/pipelines.go +++ b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/pipelines.go @@ -151,6 +151,8 @@ type pipelineEnds struct { // 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 + // runEnded is told a run on request's group-ended, its group `run-`, to drop the run's candidate (A-235); nil drops nothing. + runEnded func(group string) error // onBranches is told the branches a registration names, for the catalog half of each to be followed (design 47 G §14, §195). onBranches func([]string) // trunk names the first watched branch, the trunk's (design 47 G §507), the branch of a module registered with none. @@ -251,7 +253,8 @@ func (e *pipelineEnds) hearPipeline(m jetstream.Msg) { _ = m.Ack() } -// hearGroup takes a group-ended: each member's end, passed only if the group passed. +// hearGroup takes a group-ended: each member's end, passed only if the group passed. A run on request's (group `run-`) delivers +// nothing: it drops the run's candidate assignment graph (A-235). func (e *pipelineEnds) hearGroup(m jetstream.Msg) { g, err := mprc.GroupEnded(m.Data()) if err != nil { @@ -259,6 +262,17 @@ func (e *pipelineEnds) hearGroup(m jetstream.Msg) { _ = m.Ack() return } + if strings.HasPrefix(g.Group, runGroupPrefix) { + if e.runEnded != nil { + if err := e.runEnded(g.Group); err != nil { + e.logf("%s: %v; it is heard again", m.Subject(), err) + _ = m.Nak() + return + } + } + _ = 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) 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 d0c6fa1c..801f4d10 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 @@ -21,6 +21,7 @@ import ( oc "git.novox.be/novox/mesh-contracts/operator-and-conversation/go/seats/operatorchannel" "git.novox.be/novox/mesh-sdk/go/catalogue" + "github.com/nats-io/nats.go/jetstream" "git.novox.be/novox/mesh-delivery/modules/mesh-grants/grantsstate" @@ -54,6 +55,19 @@ func (m *memEnds) swap(key string, v []byte, rev uint64) error { return nil } +// heardMsg is one message as a handler of the hub is handed it, saying whether it was acknowledged or given back. +type heardMsg struct { + jetstream.Msg + data []byte + subject string + acked, naked bool +} + +func (m *heardMsg) Data() []byte { return m.data } +func (m *heardMsg) Subject() string { return m.subject } +func (m *heardMsg) Ack() error { m.acked = true; return nil } +func (m *heardMsg) Nak() error { m.naked = true; 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, Fingerprint: fp, Outcome: outcome, Steps: steps, Pipeline: branch + "." + module, Commit: "c"} @@ -111,6 +125,35 @@ func TestAWantFollowsAPassedPipeline(t *testing.T) { t.Errorf("%s: may want = %v, want %v", c.name, got, c.may) } } + // A run on request's group-ended (group run-, A-235) writes no want and drops the run's candidate; a drop that fails is + // heard again. + { + e := newPipelineEnds(newMemEnds(), func() {}, t.Logf) + e.on.Store(true) + var dropped []string + fail := false + e.runEnded = func(group string) error { + dropped = append(dropped, group) + if fail { + return fmt.Errorf("the bucket is not open") + } + return nil + } + raw, _ := json.Marshal(mpr.GroupEnded{V: 1, Branch: "main", Group: "run-7", Outcome: mpr.GroupEndedOutcomePassed, + Pipelines: []mpr.PipelineEnded{func() mpr.PipelineEnded { p := pipe("web", "main", 5, 1, fp1, pass); p.Run = "7"; return p }()}}) + m := &heardMsg{data: raw, subject: "mesh.seat.mesh-pipeline-runner.event.group-ended.x"} + e.hearGroup(m) + if e.mayWant("web", fp1, []string{"main"}, nil) || !slices.Equal(dropped, []string{"run-7"}) || !m.acked { + t.Errorf("a run's group-ended: may want %v, dropped %v, acked %v; want no want, run-7 dropped, acked", + e.mayWant("web", fp1, []string{"main"}, nil), dropped, m.acked) + } + fail = true + m = &heardMsg{data: raw, subject: m.subject} + e.hearGroup(m) + if !m.naked { + t.Error("a run's candidate not dropped is not heard again") + } + } // 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) diff --git a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/views.go b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/views.go index 8391bf3f..4eb76b26 100644 --- a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/views.go +++ b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/views.go @@ -3,15 +3,19 @@ package main import ( "context" "encoding/json" + "fmt" "mesh-delivery-assigner/internal/assignment" "mesh-delivery-assigner/internal/buckets" "slices" + "sync/atomic" "time" "github.com/nats-io/nats.go/jetstream" mma "git.novox.be/novox/mesh-contracts/go/seats/meshmoduleassigner" "git.novox.be/novox/mesh-contracts/health-and-repair/go/clients/meshhandactsclient" + mcc "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/clients/meshcatalogclient" + "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/seats/meshcatalog" stdio "git.novox.be/novox/mesh-sdk/go" "git.novox.be/novox/mesh-sdk/go/process" @@ -122,6 +126,9 @@ type catalogViews struct { hub *process.EventHub h *halves + // candidates is told each candidate's catalog-revision (A-201), once the passer that keeps its assignment graph is made; + // before that a candidate's revision is said not taken. + candidates atomic.Pointer[func(context.Context, meshcatalog.CatalogRevisionEvent) error] // answers is the picture of the teller's answers (design 47 G A-175, A-178): its first view through teller_read_answers, // then its views family; nil when it is not followed. answers *views.Picture @@ -142,6 +149,15 @@ func (iv *catalogViews) start(ctx context.Context, w *world, wake func(), settle if err := iv.h.follow(); err != nil { iv.logf("the catalog half is not followed, so no module is placed until the assigner starts again: %v", err) } + // Each candidate's catalog graph revision, heard on the reader's one subscription of catalog-revision (A-201, A-185). + if err := mcc.WireCandidatesProcess(reader, func(ctx context.Context, ev meshcatalog.CatalogRevisionEvent) error { + if f := iv.candidates.Load(); f != nil { + return (*f)(ctx, ev) + } + return fmt.Errorf("the candidate %s's revision %d is heard before the assigner can keep its assignment graph", ev.Group, ev.Revision) + }); err != nil { + iv.logf("no candidate's catalog graph revision is heard, so no run's assignment graph is kept until the assigner starts again: %v", err) + } // The teller's answers to the assigner's asks, followed as every owner's state is (A-175), never its bucket. if pic, err := reader.Follow(buckets.TellerModule, buckets.AnswersState, buckets.TellerModule+"."+buckets.AnswersReadTool); err != nil { iv.logf("the teller's answers are not followed, so no ask is answered until the assigner starts again: %v", err) diff --git a/modules/mesh-delivery-assigner/internal/assigngraph/candidate.go b/modules/mesh-delivery-assigner/internal/assigngraph/candidate.go new file mode 100644 index 00000000..1aafe268 --- /dev/null +++ b/modules/mesh-delivery-assigner/internal/assigngraph/candidate.go @@ -0,0 +1,138 @@ +package assigngraph + +import ( + "context" + "encoding/json" + "errors" + "fmt" + "strings" + "time" + + "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/clients/meshcatalogclient" + "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/seats/meshwants" +) + +// A candidate's assignment graph (design 47 G §189, §190, A-201, A-235): computed per run on request from the candidate's catalog +// graph revision, kept apart from trunk's under the seat's key form candidate.., the group `run-`, and dropped on +// the runner's group-ended of that run; it writes no want, and the gate's heads-replaced drops nothing. + +// CandidateObject is the object name of a candidate's revision (the seat's key form candidate..). +func CandidateObject(group, heads string) string { + return meshwants.AssignmentRevisionsByCandidateGroupHeadsKey(group, heads) +} + +// KeepCandidate keeps c as the candidate's revision of group and heads, numbered after the one kept of it, unless that one lists +// the same entries from the same inputs; its entry objects are written first, an entry trunk shares stored once. +func KeepCandidate(ctx context.Context, o Objects, c Computed, group, heads, cause string, at time.Time) (meshwants.AssignmentRevision, bool, error) { + if group == "" || heads == "" { + return meshwants.AssignmentRevision{}, false, fmt.Errorf("a candidate's revision names its group and its heads, not %q and %q", group, heads) + } + was, err := readCandidateObject(ctx, o, CandidateObject(group, heads)) + has := err == nil + if err != nil && !errors.Is(err, ErrNoObject) { + return was, false, err + } + if has && Same(was, c.Revision) { + return was, false, nil + } + rev := c.Revision + rev.Kind, rev.Group, rev.Heads, rev.Cause = meshcatalogclient.KindCandidate, group, heads, cause + rev.Revision = was.Revision + 1 + t := at.UTC() + rev.At = &t + if err := writeEntries(ctx, o, c, rev); err != nil { + return was, false, err + } + list, err := json.Marshal(rev) + if err != nil { + return was, false, err + } + if err := o.Put(ctx, CandidateObject(group, heads), list); err != nil { + return was, false, fmt.Errorf("the candidate %s's revision in %s: %w", group, RevisionsObjects, err) + } + return rev, true, nil +} + +// ReadCandidate is the candidate's revision kept of group, of heads when named, else the newest kept of the group; ErrNoCandidate +// when none is kept. +func ReadCandidate(ctx context.Context, o Objects, group, heads string) (meshwants.AssignmentRevision, error) { + if heads != "" { + rev, err := readCandidateObject(ctx, o, CandidateObject(group, heads)) + if errors.Is(err, ErrNoObject) { + err = fmt.Errorf("%w (group %q, heads %q)", ErrNoCandidate, group, heads) + } + return rev, err + } + names, err := CandidateNames(ctx, o, group) + if err != nil { + return meshwants.AssignmentRevision{}, err + } + var newest meshwants.AssignmentRevision + found := false + for _, n := range names { + rev, err := readCandidateObject(ctx, o, n) + if err != nil { + return rev, err + } + if !found || rev.At != nil && newest.At != nil && rev.At.After(*newest.At) { + newest, found = rev, true + } + } + if !found { + return newest, fmt.Errorf("%w (group %q)", ErrNoCandidate, group) + } + return newest, nil +} + +// CandidateNames is the object name of every revision kept of the candidate group; every candidate's when group is "". +func CandidateNames(ctx context.Context, o Objects, group string) ([]string, error) { + names, err := o.Names(ctx) + if err != nil { + return nil, fmt.Errorf("the objects %s: %w", RevisionsObjects, err) + } + var out []string + for _, n := range names { + g, _, ok := meshwants.ParseAssignmentRevisionsByCandidateGroupHeadsKey(n) + if ok && (group == "" || g == group) { + out = append(out, n) + } + } + return out, nil +} + +// Remover is an object bucket that also removes an object. +type Remover interface { + Objects + // Delete removes the object of that name; one not held is no error. + Delete(ctx context.Context, name string) error +} + +// DropCandidate removes every revision kept of the candidate group (its run's group-ended, A-235); the entries it alone listed are +// removed with the next revisions nothing pins. It says how many it removed. +func DropCandidate(ctx context.Context, o Remover, group string) (int, error) { + if strings.TrimSpace(group) == "" { + return 0, errors.New("a candidate to drop names its group") + } + names, err := CandidateNames(ctx, o, group) + if err != nil { + return 0, err + } + for i, n := range names { + if err := o.Delete(ctx, n); err != nil { + return i, fmt.Errorf("the candidate %s's revision %s: %w", group, n, err) + } + } + return len(names), nil +} + +func readCandidateObject(ctx context.Context, o Objects, name string) (meshwants.AssignmentRevision, error) { + var rev meshwants.AssignmentRevision + data, err := o.Get(ctx, name) + if err != nil { + return rev, err + } + if err := json.Unmarshal(data, &rev); err != nil { + return rev, fmt.Errorf("the candidate's revision %s: %w", name, err) + } + return rev, nil +} diff --git a/modules/mesh-delivery-assigner/internal/assigngraph/graph_test.go b/modules/mesh-delivery-assigner/internal/assigngraph/graph_test.go index 2827b7f6..8e8b3589 100644 --- a/modules/mesh-delivery-assigner/internal/assigngraph/graph_test.go +++ b/modules/mesh-delivery-assigner/internal/assigngraph/graph_test.go @@ -236,3 +236,77 @@ func TestARevisionOf120ModulesOnFourNodesFitsOneValue(t *testing.T) { t.Errorf("%d entries in %d bytes", len(c.Revision.Entries), len(raw)) } } + +func (m memObjects) Delete(_ context.Context, name string) error { + delete(m, name) + return nil +} + +// A-201, A-235, §186, §189: a run's candidate assignment graph is kept apart from trunk's under candidate.., served by +// revision, entry and difference against trunk, and dropped by its group alone. +func TestARunsCandidateGraphIsKeptApartServedAndDropped(t *testing.T) { + ctx := context.Background() + at := time.Date(2026, 11, 12, 9, 0, 0, 0, time.UTC) + placed := map[string]mma.Assignments{"ace": {Modules: []mma.PlacedModule{{Module: "web"}}}} + first := func(string) string { return "main" } + o := memObjects{} + trunk := Compute(Inputs{Halves: map[string]meshcatalogclient.Half{"main": half("main", 3, meshcatalog.CatalogEntry{Module: "web", Fp: "src2:w1"})}, + Placements: placed, FirstBranch: first}) + if _, _, err := Keep(ctx, o, trunk, "catalog-revision", at); err != nil { + t.Fatal(err) + } + cand := Compute(Inputs{Halves: map[string]meshcatalogclient.Half{"main": half("main", 4, meshcatalog.CatalogEntry{Module: "web", Fp: "src2:w2"})}, + Placements: placed, FirstBranch: first}) + if _, _, err := KeepCandidate(ctx, o, cand, "", "h1", "catalog-revision", at); err == nil { + t.Error("a candidate naming no group was kept") + } + rev, fresh, err := KeepCandidate(ctx, o, cand, "run-7", "h1", "catalog-revision", at) + if err != nil || !fresh || rev.Kind != meshcatalogclient.KindCandidate || rev.Group != "run-7" || rev.Revision != 1 { + t.Fatalf("kept %+v fresh %v: %v", rev, fresh, err) + } + if _, ok := o[CandidateObject("run-7", "h1")]; !ok { + t.Fatalf("the candidate is not under candidate.run-7.h1: %v", o) + } + if _, fresh, _ := KeepCandidate(ctx, o, cand, "run-7", "h1", "catalog-revision", at); fresh { + t.Error("the same candidate kept again is a new revision") + } + newest, has, err := Newest(ctx, o) + if err != nil || !has || newest.Kind != meshcatalogclient.KindTrunk || newest.Revision != 1 { + t.Errorf("trunk's newest is %+v (%v, %v): the candidate is not kept apart", newest, has, err) + } + for _, in := range []meshwants.RevisionInput{{Group: "run-7"}, {Group: "run-7", Heads: "h1"}} { + a, err := Revision(ctx, o, in) + if err != nil || a.Revision.Group != "run-7" { + t.Errorf("revision %+v answered %+v: %v", in, a.Revision, err) + } + } + if _, err := Revision(ctx, o, meshwants.RevisionInput{Group: "run-8"}); !errors.Is(err, ErrNoCandidate) { + t.Errorf("an unkept candidate answered %v", err) + } + e, err := Entry(ctx, o, meshwants.EntryInput{Group: "run-7", Node: "ace", Module: "web"}) + if err != nil || e.Entry.Fp != "src2:w2" { + t.Errorf("the candidate's entry is %+v: %v", e.Entry, err) + } + d, err := Difference(ctx, o, meshwants.DifferenceInput{Group: "run-7"}) + if err != nil || len(d.Changes) != 1 || d.Changes[0].From != "src2:w1" || d.Changes[0].To != "src2:w2" { + t.Errorf("the candidate against trunk is %+v: %v", d, err) + } + if _, err := Difference(ctx, o, meshwants.DifferenceInput{}); err == nil { + t.Error("difference with none of from, to and group is not refused") + } + if _, _, err := KeepCandidate(ctx, o, cand, "run-9", "h2", "catalog-revision", at); err != nil { + t.Fatal(err) + } + if n, err := DropCandidate(ctx, o, "run-7"); err != nil || n != 1 { + t.Errorf("dropped %d: %v", n, err) + } + if _, err := ReadCandidate(ctx, o, "run-7", ""); !errors.Is(err, ErrNoCandidate) { + t.Errorf("the dropped candidate reads: %v", err) + } + if _, err := ReadCandidate(ctx, o, "run-9", ""); err != nil { + t.Errorf("another run's candidate went with it: %v", err) + } + if _, has, _ := Newest(ctx, o); !has { + t.Error("dropping a candidate dropped trunk") + } +} diff --git a/modules/mesh-delivery-assigner/internal/assigngraph/keep.go b/modules/mesh-delivery-assigner/internal/assigngraph/keep.go index 7ce8600e..743d346e 100644 --- a/modules/mesh-delivery-assigner/internal/assigngraph/keep.go +++ b/modules/mesh-delivery-assigner/internal/assigngraph/keep.go @@ -103,36 +103,13 @@ func Keep(ctx context.Context, o Objects, c Computed, cause string, at time.Time if has && Same(newest, c.Revision) { return newest, false, nil } - names, err := o.Names(ctx) - if err != nil { - return newest, false, fmt.Errorf("the objects %s: %w", RevisionsObjects, err) - } - held := make(map[string]bool, len(names)) - for _, n := range names { - held[n] = true - } rev := c.Revision rev.Revision = newest.Revision + 1 rev.Cause = cause t := at.UTC() rev.At = &t - for _, ref := range rev.Entries { - name := EntryObject(ref.Digest) - if held[name] { - continue - } - e, ok := c.Entries[ref.Digest] - if !ok { - return newest, false, fmt.Errorf("revision %d lists the entry of %s/%s (%s), which it does not carry", rev.Revision, ref.Node, ref.Module, ref.Digest) - } - data, err := json.Marshal(e) - if err != nil { - return newest, false, err - } - if err := o.Put(ctx, name, data); err != nil { - return newest, false, fmt.Errorf("the entry of %s/%s in %s: %w", ref.Node, ref.Module, RevisionsObjects, err) - } - held[name] = true + if err := writeEntries(ctx, o, c, rev); err != nil { + return newest, false, err } list, err := json.Marshal(rev) if err != nil { @@ -144,6 +121,37 @@ func Keep(ctx context.Context, o Objects, c Computed, cause string, at time.Time return rev, true, nil } +// writeEntries writes each entry object rev lists that the bucket does not hold yet, from c, before its list is kept. +func writeEntries(ctx context.Context, o Objects, c Computed, rev meshwants.AssignmentRevision) error { + names, err := o.Names(ctx) + if err != nil { + return fmt.Errorf("the objects %s: %w", RevisionsObjects, err) + } + held := make(map[string]bool, len(names)) + for _, n := range names { + held[n] = true + } + for _, ref := range rev.Entries { + name := EntryObject(ref.Digest) + if held[name] { + continue + } + e, ok := c.Entries[ref.Digest] + if !ok { + return fmt.Errorf("revision %d lists the entry of %s/%s (%s), which it does not carry", rev.Revision, ref.Node, ref.Module, ref.Digest) + } + data, err := json.Marshal(e) + if err != nil { + return err + } + if err := o.Put(ctx, name, data); err != nil { + return fmt.Errorf("the entry of %s/%s in %s: %w", ref.Node, ref.Module, RevisionsObjects, err) + } + held[name] = true + } + return nil +} + // Event is the seat's event assignment-revision of a kept revision. func Event(rev meshwants.AssignmentRevision) meshwants.AssignmentRevisionEvent { return meshwants.AssignmentRevisionEvent{V: rev.V, Revision: rev.Revision, Kind: rev.Kind, Cause: rev.Cause, Group: rev.Group, diff --git a/modules/mesh-delivery-assigner/internal/assigngraph/verbs.go b/modules/mesh-delivery-assigner/internal/assigngraph/verbs.go index 3b7d2d6c..525fee58 100644 --- a/modules/mesh-delivery-assigner/internal/assigngraph/verbs.go +++ b/modules/mesh-delivery-assigner/internal/assigngraph/verbs.go @@ -10,16 +10,23 @@ import ( "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/seats/meshwants" ) -// ErrNoCandidate answers a verb asked of a candidate's revision: the assigner keeps none (the candidate's graph and the -// pins from the gate are milestone 47/G's W5.9, held with research 086's fold). -var ErrNoCandidate = errors.New("the assigner keeps no candidate's assignment graph revision") +// ErrNoCandidate answers a verb asked of a candidate's revision the assigner keeps none of: a run's candidate is kept from its +// catalog graph revision until its run's group-ended (A-201, A-235). +var ErrNoCandidate = errors.New("the assigner keeps no such candidate's assignment graph revision") // ErrNoRevision answers a verb asked while no trunk revision is kept yet. var ErrNoRevision = errors.New("no assignment graph revision is kept yet") func revisionOf(ctx context.Context, o Objects, revision int64, group, heads string) (meshwants.AssignmentRevision, error) { if group != "" || heads != "" { - return meshwants.AssignmentRevision{}, fmt.Errorf("%w (group %q, heads %q)", ErrNoCandidate, group, heads) + if group == "" { + return meshwants.AssignmentRevision{}, fmt.Errorf("a candidate's revision is asked by its group, with or without its heads, not by heads %q alone", heads) + } + rev, err := ReadCandidate(ctx, o, group, heads) + if err == nil && revision > 0 && rev.Revision != revision { + err = fmt.Errorf("%w (group %q, revision %d: the one kept is %d)", ErrNoCandidate, group, revision, rev.Revision) + } + return rev, err } if revision > 0 { return ReadRevision(ctx, o, revision) @@ -66,7 +73,16 @@ func Entry(ctx context.Context, o Objects, in meshwants.EntryInput) (meshwants.E // differs, with the edges that moved, and the nodes it touches. func Difference(ctx context.Context, o Objects, in meshwants.DifferenceInput) (meshwants.Difference, error) { if in.Group != "" || in.Heads != "" { - return meshwants.Difference{}, fmt.Errorf("%w (group %q, heads %q)", ErrNoCandidate, in.Group, in.Heads) + // A candidate against trunk (§186, §187): from the trunk revision named, else the newest kept. + to, err := revisionOf(ctx, o, 0, in.Group, in.Heads) + if err != nil { + return meshwants.Difference{}, err + } + from, err := revisionOf(ctx, o, in.From, "", "") + if err != nil { + return meshwants.Difference{}, err + } + return Compare(ctx, o, from, to) } if in.From <= 0 || in.To <= 0 { return meshwants.Difference{}, errors.New("difference takes the two kept revisions it compares, from and to")