jschoubben
02f50bd9d6
Since the build CPU cap (issue 1806) shanks' mesh-build-go kept capacity 3 inside a 12-core slice: three full checks got about 4 cores each and ran past the 30-minute step bound (issue 1890). The puller now judges by min(configured, floor(slice quota cores / cores-per-step)), the quota read live from its cgroup, says it on change and writes capacity.<node> to the pauses state. A run's phases (fetch and merge, digest, tests) are logged with the ask id and written to its running step key; Asker.Wait hands every step value to a WithStepSeen hook.
68 lines
2.0 KiB
Go
68 lines
2.0 KiB
Go
package buildstep
|
|
|
|
import (
|
|
"context"
|
|
"strings"
|
|
"sync"
|
|
"time"
|
|
|
|
"git.novox.be/novox/mesh-sdk/go/delivery"
|
|
)
|
|
|
|
// **A running ask says its phases** (issue 1890, 2026-10-26 19:09: a mesh-catalog try's step on shanks ran 30 minutes
|
|
// and its module's journal held no line naming its ask, so nobody could tell whether it fetched, linted or tested). The
|
|
// puller hands every ask's run a phase reporter in its context; RunAsk and a module's own step bodies call Phase as they
|
|
// move (fetch and merge, a level's digest, lint, a level's tests), and each phase is logged with the ask's id and its
|
|
// time since the start, and written to the ask's running step key (delivery.Step.Phase), so its asker and the delivery
|
|
// view read where it stands from the bus.
|
|
|
|
type phaseKey struct{}
|
|
|
|
// Phase says that the ask whose run ctx belongs to entered phase; nothing outside an ask's run (a tool called by hand).
|
|
func Phase(ctx context.Context, phase string) {
|
|
if r, ok := ctx.Value(phaseKey{}).(*phases); ok && r != nil {
|
|
r.enter(strings.TrimSpace(phase))
|
|
}
|
|
}
|
|
|
|
// phases is one running ask's reporter: it writes the step key again with each new phase until the step ends.
|
|
type phases struct {
|
|
mu sync.Mutex
|
|
put Put
|
|
key string
|
|
step delivery.Step
|
|
ended bool
|
|
logf func(format string, args ...any)
|
|
prefix string
|
|
}
|
|
|
|
func (r *phases) enter(phase string) {
|
|
if phase == "" {
|
|
return
|
|
}
|
|
r.mu.Lock()
|
|
defer r.mu.Unlock()
|
|
if r.ended || phase == r.step.Phase {
|
|
return
|
|
}
|
|
at := now().UTC()
|
|
r.step.Phase, r.step.PhaseAt = phase, at
|
|
if r.logf != nil {
|
|
r.logf("%s: %s, %s after its start", r.prefix, phase, at.Sub(r.step.Started).Round(time.Second))
|
|
}
|
|
if r.put == nil {
|
|
return
|
|
}
|
|
if err := write(r.put, r.key, r.step); err != nil && r.logf != nil {
|
|
r.logf("%s: the phase %s could not be written to %s: %v", r.prefix, phase, r.key, err)
|
|
}
|
|
}
|
|
|
|
// end stops the writes (the step's end is written after it) and answers the step with its last phase, for the end.
|
|
func (r *phases) end() delivery.Step {
|
|
r.mu.Lock()
|
|
defer r.mu.Unlock()
|
|
r.ended = true
|
|
return r.step
|
|
}
|