From 0d98108bc4573fb818659316d264b461c988e25f Mon Sep 17 00:00:00 2001 From: jochen Date: Thu, 12 Nov 2026 09:54:12 +0100 Subject: [PATCH] AS5-4: a running composition step says step-progress once a minute MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A ticker inside the running step, stopped at its end, says the seat mesh-wants' step-progress. with the ask, the asker runner, the phase the run is in (fetch while its revision and inputs are read, run while its run asks run, report) and the time. Builds: design 47 G §496, §603, A-262, A-267 Claude-Session: https://claude.ai/code/session_01NcbgfGKaxw984XyAPfLm4h --- .../mesh-delivery-assigner/compositionrun.go | 2 + .../mesh-delivery-assigner/compositionstep.go | 69 ++++++++++++++++++- .../compositionstep_test.go | 48 +++++++++++++ .../cmd/mesh-delivery-assigner/main.go | 2 +- 4 files changed, 119 insertions(+), 2 deletions(-) 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) }