AS5-1: the assigner takes the composition step and says its step-ended
The seat mesh-wants' accepted kind step is taken for the composition step and its end said as step-ended.<node> under the ask's id; it is wired with its run in AS5-3. Builds: design 47 G §22, §354 Claude-Session: https://claude.ai/code/session_01NcbgfGKaxw984XyAPfLm4h
This commit is contained in:
@@ -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.<node>, 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
|
||||
}
|
||||
@@ -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.<node> 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)
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user