diff --git a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/compositionrun.go b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/compositionrun.go index edf581ca..14acf61b 100644 --- a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/compositionrun.go +++ b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/compositionrun.go @@ -156,6 +156,7 @@ func (r *compositionRun) run(ctx context.Context, _ string, run meshwants.StepRu if err != nil { return couldNotRun(err.Error()) } + setPhase(ctx, meshwants.StepProgressEventPhaseRun) ends := make([]machineEnd, len(inputs)) var wg sync.WaitGroup for i, in := range inputs { @@ -166,6 +167,7 @@ func (r *compositionRun) run(ctx context.Context, _ string, run meshwants.StepRu }() } wg.Wait() + setPhase(ctx, meshwants.StepProgressEventPhaseReport) var out stepOutcome var failed, couldNot []string for _, e := range ends { diff --git a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/compositionstep.go b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/compositionstep.go index b548d835..70949222 100644 --- a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/compositionstep.go +++ b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/compositionstep.go @@ -3,6 +3,7 @@ package main import ( "context" "encoding/json" + "sync" "time" "github.com/nats-io/nats.go" @@ -31,6 +32,40 @@ type compositionSteps struct { run func(ctx context.Context, id string, run meshwants.StepRun) stepOutcome now func() time.Time logf func(string, ...any) + // asker is the name the step's progress names as its asker. + asker string + // ticks is a ticker of every d, its channel and its stop (time.NewTicker); nil is the clock's. A test hands its own. + ticks func(d time.Duration) (<-chan time.Time, func()) +} + +// stepAsker is the asker a composition step's progress names: the pipeline runner, the one asker that sends the accepted kind +// step on mesh-wants (design 47 G A-267), whose accept subject names no asker. +const stepAsker = "runner" + +// progressEvery is how often a running composition step says its progress (design 47 G §496, A-262). +const progressEvery = time.Minute + +// stepPhase is the phase a running step is in: fetch, build, run or report (§496), set by the run through its context. +type stepPhase struct { + mu sync.Mutex + phase string +} + +type phaseKey struct{} + +// setPhase sets the phase of the step ctx runs, when it runs one. +func setPhase(ctx context.Context, phase string) { + if p, ok := ctx.Value(phaseKey{}).(*stepPhase); ok { + p.mu.Lock() + p.phase = phase + p.mu.Unlock() + } +} + +func (p *stepPhase) get() string { + p.mu.Lock() + defer p.mu.Unlock() + return p.phase } // 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 @@ -48,7 +83,10 @@ func (c *compositionSteps) take(ctx context.Context, id string, data []byte) boo // 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) + phase := &stepPhase{phase: meshwants.StepProgressEventPhaseFetch} + stop := c.sayProgress(id, phase) + o := c.run(context.WithValue(ctx, phaseKey{}, phase), id, run) + stop() 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, @@ -58,6 +96,35 @@ func (c *compositionSteps) runOne(ctx context.Context, id string, run meshwants. } } +// sayProgress says the step's progress on the seat's step-progress. once a minute until the returned stop is called (design +// 47 G §496, A-262): a ticker inside the running step, as the build seats' step runner says its own beat. +func (c *compositionSteps) sayProgress(id string, phase *stepPhase) (stop func()) { + ticks := c.ticks + if ticks == nil { + ticks = func(d time.Duration) (<-chan time.Time, func()) { t := time.NewTicker(d); return t.C, t.Stop } + } + tick, stopTicks := ticks(progressEvery) + done := make(chan struct{}) + ended := make(chan struct{}) + go func() { + defer close(ended) + defer stopTicks() + for { + select { + case <-done: + return + case <-tick: + at := c.now().UTC() + p := meshwants.StepProgressEvent{Ask: id, Asker: c.asker, Phase: phase.get(), At: at} + if err := c.say(meshwants.EventStepProgress+"."+c.node, p, id+".progress."+at.Format(time.RFC3339Nano)); err != nil { + c.logf("saying the progress of the composition step %s: %v", id, err) + } + } + } + }() + return func() { close(done); <-ended } +} + // 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 != "" { 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 index 8e40c3ee..e5c7f566 100644 --- a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/compositionstep_test.go +++ b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/compositionstep_test.go @@ -70,3 +70,51 @@ func TestTheCompositionStepIsTakenAndItsEndSaid(t *testing.T) { t.Fatalf("step-ended %+v", end) } } + +// Design 47 G §496, A-262: a running composition step says step-progress. on every tick of its ticker, with the ask, its +// asker, the phase the run is in and the time, and says none once it has ended. +func TestARunningCompositionStepSaysItsProgress(t *testing.T) { + said := newSaidEvents() + tick := make(chan time.Time) + stopped := make(chan struct{}) + at := time.Date(2026, 11, 12, 9, 0, 0, 0, time.UTC) + release := make(chan struct{}) + c := &compositionSteps{node: "n1", say: said.say, now: func() time.Time { return at }, logf: t.Logf, asker: "runner", + ticks: func(d time.Duration) (<-chan time.Time, func()) { + if d != time.Minute { + t.Errorf("the progress ticks every %s, want a minute", d) + } + return tick, func() { close(stopped) } + }, + run: func(ctx context.Context, _ string, _ meshwants.StepRun) stepOutcome { + setPhase(ctx, meshwants.StepProgressEventPhaseRun) + <-release + return stepOutcome{Outcome: meshwants.StepEndedOutcomePassed} + }} + body, _ := json.Marshal(meshwants.StepRun{Pipeline: "main.4.m", Step: meshwants.StepRunStepComposition, Module: "m", Branch: "main"}) + if !c.take(context.Background(), "ask-1", body) { + t.Fatal("the composition step is not taken") + } + progress := meshwants.EventStepProgress + ".n1" + for range 2 { + tick <- at + if e := <-said.heard; e != progress { + t.Fatalf("said %s on a tick, want %s", e, progress) + } + } + said.mu.Lock() + p, ok := said.said[progress].(meshwants.StepProgressEvent) + said.mu.Unlock() + if !ok || p.Ask != "ask-1" || p.Asker != "runner" || p.Phase != meshwants.StepProgressEventPhaseRun || !p.At.Equal(at) { + t.Fatalf("step-progress %+v", p) + } + close(release) + if e := <-said.heard; e != meshwants.EventStepEnded+".n1" { + t.Fatalf("said %s at the end, want step-ended", e) + } + select { + case <-stopped: + default: + t.Fatal("the ticker is not stopped when the step has ended") + } +} diff --git a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/main.go b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/main.go index d2599830..218d4761 100644 --- a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/main.go +++ b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/main.go @@ -641,7 +641,7 @@ func follow(ctx context.Context, js jetstream.JetStream, nc process.Subscriber, if bus != nil && js != nil && inputs != nil && p.publishWants != nil { run := &compositionRun{inputs: &compositionInputs{machines: w.machinesOf, call: bus.CallKeptOn}, keeper: inputs, call: bus.CallKept, trunk: trunkOf(iv), running: w.runningOf, away: away, put: runAskPublisher(js), ends: ends, ask: bus.Call, now: time.Now, logf: logger.Printf} - steps := &compositionSteps{node: thisNode, say: p.publishWants, run: run.run, now: time.Now, logf: logger.Printf} + steps := &compositionSteps{node: thisNode, say: p.publishWants, run: run.run, now: time.Now, logf: logger.Printf, asker: stepAsker} if err := serveCompositionSteps(ctx, nc, steps); err != nil { logger.Printf("the composition step is not heard, so no composition step runs until the assigner starts again: %v", err) }