AS12: the assigner keeps a run's candidate assignment graph, dropped on the run's group-ended

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.<group>.<heads>, the group run-<id>. 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
This commit is contained in:
2026-11-12 09:14:45 +01:00
parent 019a7ab9f7
commit d5676ee4b3
13 changed files with 543 additions and 39 deletions
@@ -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-<id>`'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.<group>.<heads>;
// 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-<id>` (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
}
@@ -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 {
@@ -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.<group>.<heads> 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)
}
}
@@ -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)
@@ -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] {
@@ -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" {
@@ -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-<id>`, 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-<id>`) 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)
@@ -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-<id>, 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)
@@ -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)
@@ -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.<group>.<heads>, the group `run-<id>`, 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.<group>.<heads>).
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
}
@@ -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.<group>.<heads>, 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")
}
}
@@ -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,
@@ -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")