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 e7a555e8..8c1eef61 100644 --- a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/pipelines.go +++ b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/pipelines.go @@ -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 { 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 f2e0a12b..535fa85c 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 @@ -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",