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 }