From c68d51b8b58dffc583be70aab45af5e6b1eb6962 Mon Sep 17 00:00:00 2001 From: jochen Date: Fri, 13 Nov 2026 02:16:00 +0100 Subject: [PATCH] Assigner (A-334): in a group that changes the runner, the runner's member is wanted on its own end MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit In a group take-in that changes the runner, the other members' pipelines are held for the new runner, so the group never ended before the bound and the new runner was never wanted. The runner is the acting part the others wait on, so it goes first as the node-engine does: its member (the holder of the seat mesh-pipeline-runner, read from the acting edges) is wanted on its own pipeline-ended once that passed, ahead of the group; the other members keep the group's end. Builds: design 47 G §29, §65, §268; A-328, A-334 Claude-Session: https://claude.ai/code/session_01NcbgfGKaxw984XyAPfLm4h --- .../cmd/mesh-delivery-assigner/graphkept.go | 4 +++ .../cmd/mesh-delivery-assigner/pipelines.go | 21 +++++++++++-- .../mesh-delivery-assigner/pipelines_test.go | 31 +++++++++++++++++++ .../internal/assigngraph/cycle.go | 14 +++++++++ 4 files changed, 67 insertions(+), 3 deletions(-) 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 0040b801..d8a20de6 100644 --- a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/graphkept.go +++ b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/graphkept.go @@ -6,6 +6,7 @@ import ( "encoding/hex" "encoding/json" "fmt" + "maps" "slices" "strings" "time" @@ -52,6 +53,9 @@ func (p *passer) keepGraph(ctx context.Context) { } c := assigngraph.Compute(p.graphInputs()) g.computed = c + if p.ends != nil { + p.ends.setRunner(assigngraph.Runner(slices.Collect(maps.Values(c.Entries)))) // the runner's member of a group goes first (A-334) + } for _, u := range c.Uncomputed { if g.said != u { g.said = u 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 cedaedfa..e7a555e8 100644 --- a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/pipelines.go +++ b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/pipelines.go @@ -157,6 +157,19 @@ type pipelineEnds struct { onBranches func([]string) // trunk names the first watched branch, the trunk's (design 47 G §507), the branch of a module registered with none. trunk func() string + // runner is the module holding the seat mesh-pipeline-runner, as the last pass read it from the acting edges (A-334). + runner atomic.Pointer[string] +} + +// setRunner keeps the module holding the seat mesh-pipeline-runner, read by the pass from the acting edges (A-334). +func (e *pipelineEnds) setRunner(module string) { + e.runner.Store(&module) +} + +// isRunner says whether module holds the seat mesh-pipeline-runner, as the last pass read it. +func (e *pipelineEnds) isRunner(module string) bool { + r := e.runner.Load() + return r != nil && *r != "" && *r == module } func newPipelineEnds(st endStore, wake func(), logf func(string, ...any)) *pipelineEnds { @@ -235,14 +248,16 @@ func (e *pipelineEnds) branchOf(module, placed string) string { return "" } -// 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). +// hearPipeline takes a pipeline-ended. A pipeline of a group writes nothing on its own end, but the runner's: in a group +// take-in that changes the runner it is the acting part the other members wait on, so it goes first, wanted on its own end +// once that passed, ahead of the group (A-334, as the node-engine does, §268). 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 != "": + case p.Run != "" || (p.Group != "" && !e.isRunner(p.Module)): default: if err := e.take(p, 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 f2476c4e..f2e0a12b 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 @@ -28,7 +28,10 @@ import ( "git.novox.be/novox/mesh-contracts/go/clients/nodeengineclient" "git.novox.be/novox/mesh-contracts/go/seats/nodeengine" + "mesh-delivery-assigner/internal/assigngraph" "mesh-delivery-assigner/internal/buckets" + + mws "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/seats/meshwants" ) // memEnds is endStore over a map, with revisions. @@ -154,6 +157,34 @@ func TestAWantFollowsAPassedPipeline(t *testing.T) { t.Error("a run's candidate not dropped is not heard again") } } + // A-334 (§29, §65, §268): in a group take-in that changes the runner, the runner's member (the holder of the seat + // mesh-pipeline-runner, read from the acting edges) is wanted on its own pipeline-ended once it passed, ahead of the group; + // the other members wait for group-ended. + { + e := newPipelineEnds(newMemEnds(), func() {}, t.Logf) + e.on.Store(true) + e.setRunner(assigngraph.Runner([]mws.AssignmentEntry{{Node: "ace", Module: "web", Edges: []mws.Edge{ + {Kind: assigngraph.EdgeActing, Node: "ace", Module: "runner", Seat: mpr.Seat, Step: "pipeline"}}}})) + hear := func(p mpr.PipelineEnded) { + p.Group = "g" + raw, _ := json.Marshal(p) + m := &heardMsg{data: raw, subject: "mesh.seat.mesh-pipeline-runner.event.pipeline-ended.x"} + e.hearPipeline(m) + if !m.acked { + t.Errorf("the pipeline-ended of %s is not acknowledged", p.Module) + } + } + hear(pipe("runner", "main", 5, 1, "fp-runner", pass, step(req, "passed"))) + hear(pipe("web", "main", 5, 1, fp1, pass, step(req, "passed"))) + if !e.mayWant("runner", "fp-runner", []string{"main"}, nil) || e.mayWant("web", fp1, []string{"main"}, nil) { + t.Errorf("a group changing the runner: may want the runner %v, web %v; want the runner ahead of the group and web not yet", + e.mayWant("runner", "fp-runner", []string{"main"}, nil), e.mayWant("web", fp1, []string{"main"}, nil)) + } + hear(pipe("runner", "main", 6, 1, "fp-runner-2", fail, step(req, "failed"))) + if e.mayWant("runner", "fp-runner-2", []string{"main"}, nil) { + t.Error("the runner's member that failed is wanted") + } + } // 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/internal/assigngraph/cycle.go b/modules/mesh-delivery-assigner/internal/assigngraph/cycle.go index 014da0e6..d95014f9 100644 --- a/modules/mesh-delivery-assigner/internal/assigngraph/cycle.go +++ b/modules/mesh-delivery-assigner/internal/assigngraph/cycle.go @@ -7,6 +7,7 @@ import ( "strings" "git.novox.be/novox/mesh-contracts/go/seats/nodeengine" + "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/seats/meshpipelinerunner" "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/seats/meshwants" ) @@ -102,6 +103,19 @@ func Engines(entries []meshwants.AssignmentEntry) map[string]string { return out } +// Runner is the module that holds the seat mesh-pipeline-runner, read from the acting edges' seat as Engines reads the node-engine +// (A-334); "" when none is read. +func Runner(entries []meshwants.AssignmentEntry) string { + for _, e := range entries { + for _, edge := range e.Edges { + if edge.Kind == EdgeActing && edge.Seat == meshpipelinerunner.Seat { + return edge.Module + } + } + } + return "" +} + // hop is one edge of a cycle: from the entry at, in step, to the next hop's entry. type hop struct{ at, step string }