Assigner (A-334): hold a group member's end until the graph says which module is the runner

The runner was learnt in memory at the first pass, so a group's runner
pipeline-ended heard before it was acknowledged and dropped, and the runner
was never wanted ahead of its group. Such an end is now kept unacknowledged
until a pass reads the runner from the acting edges, then taken.

Builds: design 47 G §29, §65, §268, A-334

Claude-Session: https://claude.ai/code/session_01NcbgfGKaxw984XyAPfLm4h
This commit is contained in:
2026-11-13 02:30:13 +01:00
parent 8ef266ad6f
commit 78cf4e7666
2 changed files with 52 additions and 8 deletions
@@ -157,13 +157,44 @@ 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 is the module holding the seat mesh-pipeline-runner, as the last pass read it from the acting edges (A-334); nil
// until a pass read the graph.
runner atomic.Pointer[string]
// held are the group members' pipeline-ended heard before a pass read the graph, by their body: unacknowledged, so the
// bus keeps them, and taken once the runner is read (A-334), never dropped for the runner not being known yet.
held map[string]jetstream.Msg
}
// setRunner keeps the module holding the seat mesh-pipeline-runner, read by the pass from the acting edges (A-334).
// setRunner keeps the module holding the seat mesh-pipeline-runner, read by the pass from the acting edges (A-334); the
// first one takes the group members' ends held until the graph was read.
func (e *pipelineEnds) setRunner(module string) {
e.runner.Store(&module)
e.mu.Lock()
first := e.runner.Swap(&module) == nil
held := e.held
e.held = nil
e.mu.Unlock()
if !first {
return
}
for _, m := range held {
e.hearPipeline(m)
}
}
// holdUntilRead keeps m, a group member's pipeline-ended, until a pass reads the runner from the graph; false when the
// runner is read already, so m is taken now.
func (e *pipelineEnds) holdUntilRead(m jetstream.Msg) bool {
e.mu.Lock()
defer e.mu.Unlock()
if e.runner.Load() != nil {
return false
}
if e.held == nil {
e.held = map[string]jetstream.Msg{}
}
e.held[string(m.Data())] = m
e.logf("%s is a group member's end heard before the assignment graph is read: held until a pass reads which module is the runner (A-334)", m.Subject())
return true
}
// isRunner says whether module holds the seat mesh-pipeline-runner, as the last pass read it.
@@ -257,6 +288,8 @@ func (e *pipelineEnds) hearPipeline(m jetstream.Msg) {
switch {
case err != nil:
e.logf("%s is not taken: %v", m.Subject(), err)
case p.Run == "" && p.Group != "" && e.holdUntilRead(m):
return
case p.Run != "" || (p.Group != "" && !e.isRunner(p.Module)):
default:
if err := e.take(p, requiredPassed(p)); err != nil {
@@ -163,18 +163,29 @@ func TestAWantFollowsAPassedPipeline(t *testing.T) {
{
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) {
heard := func(p mpr.PipelineEnded) *heardMsg {
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 {
return m
}
hear := func(p mpr.PipelineEnded) {
if m := heard(p); !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")))
// The runner's end heard before a pass read the graph is held, unacknowledged, and taken once the runner is read.
early := heard(pipe("runner", "main", 5, 1, "fp-runner", pass, step(req, "passed")))
if early.acked || early.naked || e.mayWant("runner", "fp-runner", []string{"main"}, nil) {
t.Fatalf("the runner's end before the graph is read: acked %v, naked %v, wanted %v; want it held", early.acked, early.naked,
e.mayWant("runner", "fp-runner", []string{"main"}, nil))
}
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"}}}}))
if !early.acked {
t.Error("the runner's end held until the graph was read is not taken once it is")
}
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",