diff --git a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/compositionstep.go b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/compositionstep.go new file mode 100644 index 00000000..79c6a2be --- /dev/null +++ b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/compositionstep.go @@ -0,0 +1,76 @@ +package main + +import ( + "context" + "encoding/json" + "time" + + "github.com/nats-io/nats.go" + + meshwants "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/seats/meshwants" +) + +// The composition step (design 47 G §22, §354, §63): the assigner takes the seat mesh-wants' accepted kind `step`, the +// composition step of one module's pipeline that the runner asks, runs it, and says its end as the seat's event +// step-ended., the ask's id as the event's id and the event's `id`, which the runner holds the end to. + +// stepOutcome is what one composition step came to. +type stepOutcome struct { + Outcome string + Said string + Warnings []string +} + +// compositionSteps takes the composition steps and says their ends. +type compositionSteps struct { + node string + // say says one event of the seat mesh-wants (wantsPublisher). + say publisher + // run runs one step and answers what it came to. + run func(ctx context.Context, id string, run meshwants.StepRun) stepOutcome + now func() time.Time + logf func(string, ...any) +} + +// take takes one ask of the accepted kind `step` by its id and body: a step that is no composition step, or an ask with no +// id, is left with a log line; one taken is run in its own goroutine and its end said. +func (c *compositionSteps) take(ctx context.Context, id string, data []byte) bool { + var run meshwants.StepRun + if err := json.Unmarshal(data, &run); err != nil || id == "" || run.Step != meshwants.StepRunStepComposition { + c.logf("a step ask that is no composition step with an id is left: %v", err) + return false + } + go c.runOne(ctx, id, run) + return true +} + +// runOne runs one step and says its end. +func (c *compositionSteps) runOne(ctx context.Context, id string, run meshwants.StepRun) { + started := c.now().UTC() + o := c.run(ctx, id, run) + ended := c.now().UTC() + end := meshwants.StepEnded{V: 1, ID: id, Pipeline: run.Pipeline, Step: meshwants.StepEndedStepComposition, Module: run.Module, + Commit: run.Commit, Fingerprint: run.Fingerprint, Node: c.node, Outcome: o.Outcome, Said: o.Said, Warnings: o.Warnings, + Rerun: run.Rerun, By: run.By, Started: &started, Ended: &ended} + if err := c.say(meshwants.EventStepEnded+"."+c.node, end, id); err != nil { + c.logf("saying the end of the composition step %s: %v", id, err) + } +} + +// stepAskID is an ask's id as the asker names it: its event id, else its message id. +func stepAskID(m *nats.Msg) string { + if id := m.Header.Get("x-event-id"); id != "" { + return id + } + return m.Header.Get(nats.MsgIdHdr) +} + +// serveCompositionSteps hears the composition step asks on the seat mesh-wants' accept subject. +func serveCompositionSteps(ctx context.Context, conn *nats.Conn, c *compositionSteps) error { + sub, err := conn.Subscribe(meshwants.AcceptSubjectStep, func(m *nats.Msg) { c.take(ctx, stepAskID(m), m.Data) }) + if err != nil { + return err + } + go func() { <-ctx.Done(); _ = sub.Drain() }() + return nil +} diff --git a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/compositionstep_test.go b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/compositionstep_test.go new file mode 100644 index 00000000..8e40c3ee --- /dev/null +++ b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/compositionstep_test.go @@ -0,0 +1,72 @@ +package main + +import ( + "context" + "encoding/json" + "sync" + "testing" + "time" + + meshwants "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/seats/meshwants" +) + +// saidEvents keeps what a publisher says. +type saidEvents struct { + mu sync.Mutex + said map[string]any + ids map[string]string + heard chan string +} + +func newSaidEvents() *saidEvents { + return &saidEvents{said: map[string]any{}, ids: map[string]string{}, heard: make(chan string, 8)} +} + +func (s *saidEvents) say(event string, body any, id string) error { + s.mu.Lock() + s.said[event], s.ids[event] = body, id + s.mu.Unlock() + s.heard <- event + return nil +} + +// AS5 (§22, §354): the assigner takes mesh-wants' accepted kind `step` for the composition step and says the seat's +// step-ended. with the ask's id, what the step was asked at and what it came to; an ask of another step is left. +func TestTheCompositionStepIsTakenAndItsEndSaid(t *testing.T) { + said := newSaidEvents() + at := time.Date(2026, 11, 12, 9, 0, 0, 0, time.UTC) + var asked meshwants.StepRun + c := &compositionSteps{node: "n1", say: said.say, now: func() time.Time { return at }, logf: t.Logf, + run: func(_ context.Context, id string, run meshwants.StepRun) stepOutcome { + asked = run + return stepOutcome{Outcome: meshwants.StepEndedOutcomeFailed, Said: "a and b own /etc/x", Warnings: []string{"n3 asleep"}} + }} + other, _ := json.Marshal(map[string]any{"pipeline": "main.4.m", "step": "build", "module": "m"}) + if c.take(context.Background(), "ask-0", other) { + t.Fatal("a step ask of another step is taken") + } + body, _ := json.Marshal(meshwants.StepRun{Pipeline: "main.4.m", Step: meshwants.StepRunStepComposition, Module: "m", Branch: "main", + Commit: "c0ffee", Revision: 4, Fingerprint: "fp1", Rerun: true, By: "lead@n1"}) + if c.take(context.Background(), "", body) { + t.Fatal("a step ask with no id is taken") + } + if !c.take(context.Background(), "ask-1", body) { + t.Fatal("the composition step is not taken") + } + select { + case <-said.heard: + case <-time.After(5 * time.Second): + t.Fatal("no step-ended is said") + } + event := meshwants.EventStepEnded + ".n1" + end, ok := said.said[event].(meshwants.StepEnded) + if !ok || said.ids[event] != "ask-1" { + t.Fatalf("said %v under %v, want step-ended.n1 under the ask's id", said.said, said.ids) + } + if asked.Module != "m" || end.ID != "ask-1" || end.Pipeline != "main.4.m" || end.Step != meshwants.StepEndedStepComposition || + end.Module != "m" || end.Commit != "c0ffee" || end.Fingerprint != "fp1" || end.Node != "n1" || + end.Outcome != meshwants.StepEndedOutcomeFailed || end.Said != "a and b own /etc/x" || len(end.Warnings) != 1 || + !end.Rerun || end.By != "lead@n1" || end.Started == nil || end.Ended == nil { + t.Fatalf("step-ended %+v", end) + } +}