AS5-4: a running composition step says step-progress once a minute

A ticker inside the running step, stopped at its end, says the seat mesh-wants' step-progress.<node> 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
2026-11-12 09:54:12 +01:00
parent 4f7206702f
commit 0d98108bc4
4 changed files with 119 additions and 2 deletions
@@ -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 {
@@ -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.<node> 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 != "" {
@@ -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.<node> 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")
+	}
+}
@@ -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)
 		}