Merge remote-tracking branch 'origin/main' into feat/1126-nodes-compose

This commit is contained in:
2026-10-26 21:56:47 +01:00
16 changed files with 1024 additions and 26 deletions
+15
View File
@@ -70,6 +70,11 @@ func (s AskSteps) fetch(ctx context.Context, args map[string]any) (json.RawMessa
if s.Fetch == nil {
return nil, fetched{}, Refuse("this build module fetches nothing")
}
if _, merges := args["base"]; merges {
Phase(ctx, "fetch and merge")
} else {
Phase(ctx, "fetch")
}
out, err := s.Fetch(ctx, args)
if err != nil {
return nil, fetched{}, err
@@ -126,6 +131,7 @@ func (s AskSteps) build(ctx context.Context, job delivery.BuildJob) (any, error)
if s.Copy == nil {
return nil, Refuse("this build module copies no artifact (%s)", a.Artifact)
}
Phase(ctx, "copy "+a.Artifact)
out, err = s.Copy(ctx, map[string]any{"module": job.Module, "artifact": a.Artifact, "from": a.From})
} else {
if s.Build == nil {
@@ -135,6 +141,7 @@ func (s AskSteps) build(ctx context.Context, job delivery.BuildJob) (any, error)
if job.Publish {
args["publish"] = true
}
Phase(ctx, "build "+a.Artifact)
out, err = s.Build(ctx, args)
}
if err != nil {
@@ -196,6 +203,13 @@ func (s AskSteps) level(ctx context.Context, job delivery.CheckJob, level string
return args
}
test := func(args map[string]any) (bool, error) {
what := "the whole level"
if dir, _ := args["module"].(string); dir != "" {
what = dir
} else if pkgs, _ := args["packages"].([]string); len(pkgs) > 0 {
what = fmt.Sprintf("%d packages", len(pkgs))
}
Phase(ctx, fmt.Sprintf("%s: tests of %s", level, what))
out, err := s.Test(ctx, args)
if err != nil {
return false, err
@@ -287,6 +301,7 @@ func (s AskSteps) plan(ctx context.Context, job delivery.CheckJob, level string,
if job.Run != "" {
return nil, true, true
}
Phase(ctx, level+": digest")
out, err := s.Digest(ctx, withRunner(map[string]any{"workspace": f.Workspace, "level": level}))
if err != nil {
if s.Logf != nil {
+135
View File
@@ -0,0 +1,135 @@
package buildstep
import (
"encoding/json"
"fmt"
"os"
"strings"
"time"
)
// **A node may build only within its build hours** (the operator, 2026-10-26 21:18: "we should only use g14 as a builder
// outside office hours", "from 18 o'clock until 6 in the morning"). A build module's `build-hours` setting names, per
// node, the window in the node's local time in which it takes asks; outside it the capacity judged by is 0: steps running
// finish, no new one is taken. A node it does not name, or an empty setting, builds at any hour. The puller judges it at
// every judgement and wakes on one timer to the window's next edge (a clock edge, not a poll of another part).
// BuildHoursSetting is the build modules' setting of the nodes' build windows: `<node>=HH:MM-HH:MM` pairs separated by
// commas, in the node's local time; an end before its start crosses midnight (g14=18:00-06:00).
const BuildHoursSetting = "build-hours"
// BuildWindow is one node's build hours: minutes after local midnight, Start inclusive, End exclusive.
type BuildWindow struct {
Start, End int
}
// String is the window as the setting writes it.
func (w BuildWindow) String() string {
return fmt.Sprintf("%02d:%02d-%02d:%02d", w.Start/60, w.Start%60, w.End/60, w.End%60)
}
// Open says whether t, in its own location, falls in the window.
func (w BuildWindow) Open(t time.Time) bool {
m := t.Hour()*60 + t.Minute()
if w.Start < w.End {
return m >= w.Start && m < w.End
}
return m >= w.Start || m < w.End
}
// NextEdge is the first time after t, in t's location, at which the window opens or closes.
func (w BuildWindow) NextEdge(t time.Time) time.Time {
var next time.Time
for day := 0; day <= 1; day++ {
for _, m := range []int{w.Start, w.End} {
e := time.Date(t.Year(), t.Month(), t.Day()+day, m/60, m%60, 0, 0, t.Location())
if e.After(t) && (next.IsZero() || e.Before(next)) {
next = e
}
}
}
return next
}
// ParseBuildHours is node's window in a build-hours setting: nil when the setting is empty or does not name the node.
func ParseBuildHours(setting, node string) (*BuildWindow, error) {
var found *BuildWindow
for _, pair := range strings.Split(setting, ",") {
pair = strings.TrimSpace(pair)
if pair == "" {
continue
}
name, span, ok := strings.Cut(pair, "=")
from, to, ok2 := strings.Cut(span, "-")
if !ok || !ok2 || strings.TrimSpace(name) == "" {
return nil, fmt.Errorf("the build hours %q: %q is not <node>=HH:MM-HH:MM", setting, pair)
}
start, err := minuteOf(from)
if err != nil {
return nil, fmt.Errorf("the build hours %q: %v", setting, err)
}
end, err := minuteOf(to)
if err != nil {
return nil, fmt.Errorf("the build hours %q: %v", setting, err)
}
if start == end {
return nil, fmt.Errorf("the build hours %q: %s opens and closes at the same minute", setting, pair)
}
if strings.TrimSpace(name) == node {
found = &BuildWindow{Start: start, End: end}
}
}
return found, nil
}
// minuteOf reads HH:MM as minutes after midnight.
func minuteOf(s string) (int, error) {
var h, m int
s = strings.TrimSpace(s)
if n, err := fmt.Sscanf(s, "%d:%d", &h, &m); err != nil || n != 2 || h < 0 || h > 23 || m < 0 || m > 59 || len(s) != 5 {
return 0, fmt.Errorf("%q is not a time HH:MM", s)
}
return h*60 + m, nil
}
// NodeBuildHours is node's build window, read from the module's settings file at path at every call, and why it is
// none when the value cannot be read. No path, no file value or an empty one is no window and no fault. A value that
// cannot be read is no window too, so the node builds at any hour; the setting is a preference of hours, not a safety
// bound (a node that must never build has capacity 0). It is loud: the fault is said once per distinct fault through
// logf, and the puller writes it in the capacity record that build_usage shows.
func NodeBuildHours(path, node string, logf func(string, ...any)) func() (*BuildWindow, string) {
said := ""
say := func(why string) {
if logf != nil && why != said {
said = why
logf("%s, so %s builds at any hour", why, node)
}
}
fault := func(why string) (*BuildWindow, string) {
say(why)
return nil, why
}
return func() (*BuildWindow, string) {
if path == "" {
return nil, ""
}
raw, err := os.ReadFile(path)
if err != nil {
return fault(fmt.Sprintf("the setting %s cannot be read: %v", BuildHoursSetting, err))
}
var s map[string]json.RawMessage
if err := json.Unmarshal(raw, &s); err != nil {
return fault(fmt.Sprintf("the settings file %s is not JSON: %v", path, err))
}
setting := ""
if v, ok := s[BuildHoursSetting]; ok && json.Unmarshal(v, &setting) != nil {
return fault(fmt.Sprintf("the setting %s %s is not a string", BuildHoursSetting, v))
}
w, err := ParseBuildHours(setting, node)
if err != nil {
return fault(err.Error())
}
said = ""
return w, ""
}
}
+141
View File
@@ -0,0 +1,141 @@
package buildstep
import (
"encoding/json"
"fmt"
"math"
"os"
"runtime"
"strconv"
"strings"
"git.novox.be/novox/mesh-sdk/go/buildcpu"
"git.novox.be/novox/mesh-sdk/go/delivery"
)
// **A build seat's capacity follows the CPU its node gives builds** (issue 1806's follow-up; issue 1890, 2026-10-26
// 19:41: under the 50 % cap shanks' mesh-build-go kept capacity 3, three full mesh-catalog checks shared 12 cores, and
// each step ran past the 30-minute step bound with no test run). The configured capacity is an upper bound; the node's
// build slice bounds it again by the cores a step needs: floor(the slice's quota in cores / cores-per-step), one at
// least. The quota is read live from the cgroup at every judgement, so a changed cap is followed at the next take
// without a restart, and the result is said (the seat's current verb, the status tools, `capacity.<node>` in the pauses
// state that the delivery view's build_usage reads).
// CoresPerStepSetting is a build module's setting of the processors one of its steps needs: a number above zero, the
// same on every node (a step's need is the module's, not the node's). Each build module declares it with its own default
// (DefaultCoresPerStep).
const CoresPerStepSetting = "cores-per-step"
// DefaultCoresPerStep are the build modules' defaults by language, from measured steps (2026-10-26): a full mesh-catalog
// check on mesh-build-go takes what it is given (three at once took 965 % of shanks' 1200 % slice, about 4 cores each,
// and ran past the 30-minute step bound), so a Go step is given 6, half again what starved; a Node step (tsc, esbuild, a
// test runner: median 15 s) and an image build (buildah, its RUN lines mostly fetches) need 2.
var DefaultCoresPerStep = map[string]float64{delivery.AskGo: 6, delivery.AskNode: 2, delivery.AskImage: 2}
// Quota is the build slice's CPU quota as this process's cgroup holds it, in percent of one processor (1200 is 12
// cores), and where it was read: the cgroup's cpu.max (buildcpu.ReadInForce, the tighter of its unit's and its slice's),
// else the share the composer gave (buildcpu.ShareWord) over the machine's processors, else 0 and why none is known.
func Quota() (percent int, from string) {
f := buildcpu.ReadInForce()
if f.Quota > 0 {
path := f.Slice.Path
if path == "" {
path = f.Cgroup
}
return f.Quota, "the cpu.max of " + path
}
if n, err := strconv.Atoi(strings.TrimSpace(os.Getenv(buildcpu.ShareWord))); err == nil && n >= 1 && n <= 100 {
return buildcpu.Quota(n, runtime.NumCPU()), fmt.Sprintf("the share %d%% the composer gave (%s), of %d processors; no cpu.max is read", n, buildcpu.ShareWord, runtime.NumCPU())
}
return 0, "no CPU quota: neither a cpu.max on its cgroup nor a share from the composer"
}
// EffectiveCapacity is the capacity a seat's holder judges by, from its configured capacity, its node's quota (percent of
// one processor, 0 when none is known) and the cores one of its steps needs. Without a quota it is the configured one;
// with one, min(configured, floor(quota cores / coresPerStep)), and one at least while the configured one is above
// zero. A coresPerStep of zero or less is taken as 1.
func EffectiveCapacity(configured, quotaPercent int, quotaFrom string, coresPerStep float64) delivery.Capacity {
if coresPerStep <= 0 || math.IsNaN(coresPerStep) || math.IsInf(coresPerStep, 0) {
coresPerStep = 1
}
c := delivery.Capacity{Configured: configured, Effective: configured, QuotaFrom: quotaFrom, CoresPerStep: coresPerStep}
switch {
case configured <= 0:
c.Effective = 0
c.Why = fmt.Sprintf("its configured capacity is %d, so it pulls nothing", configured)
return c
case quotaPercent <= 0:
c.Why = fmt.Sprintf("its configured capacity %d; %s, so the CPU bounds nothing", configured, quotaFrom)
return c
}
c.QuotaCores = float64(quotaPercent) / 100
c.ByCPU = int(math.Floor(c.QuotaCores/coresPerStep + 1e-9))
if c.ByCPU < 1 {
c.ByCPU = 1
}
byCPU := fmt.Sprintf("%g cores of the build slice (%s) / %g cores a step = %d", c.QuotaCores, quotaFrom, coresPerStep, c.ByCPU)
if c.ByCPU < configured {
c.Effective = c.ByCPU
c.Why = fmt.Sprintf("%d, bounded by the CPU: %s, under its configured %d", c.Effective, byCPU, configured)
} else {
c.Why = fmt.Sprintf("%d, its configured capacity; the CPU allows %s", configured, byCPU)
}
return c
}
// NodeCoresPerStep is the module's cores-per-step, read from its settings file at path at every call so a change is
// followed at the next judgement; fallback when there is no path, no value or one that is not a number above zero, said
// once per distinct fault through logf.
func NodeCoresPerStep(path string, fallback float64, logf func(string, ...any)) func() float64 {
said := ""
say := func(why string) {
if logf != nil && why != said {
said = why
logf("%s, so a step is taken to need %g cores", why, fallback)
}
}
return func() float64 {
if path == "" {
return fallback
}
raw, err := os.ReadFile(path)
if err != nil {
say(fmt.Sprintf("the setting %s cannot be read: %v", CoresPerStepSetting, err))
return fallback
}
var s map[string]json.RawMessage
if err := json.Unmarshal(raw, &s); err != nil {
say(fmt.Sprintf("the settings file %s is not JSON: %v", path, err))
return fallback
}
v, ok := s[CoresPerStepSetting]
if !ok {
return fallback // not declared yet (a module before its manifest names it): the default, nothing to say
}
n, err := coresOf(v)
if err != nil {
say(fmt.Sprintf("the setting %s %s: %v", CoresPerStepSetting, v, err))
return fallback
}
said = ""
return n
}
}
// coresOf reads a cores-per-step value: a JSON number or a string holding one, above zero.
func coresOf(v json.RawMessage) (float64, error) {
var n float64
if err := json.Unmarshal(v, &n); err != nil {
var s string
if err := json.Unmarshal(v, &s); err != nil {
return 0, fmt.Errorf("is neither a number nor a string")
}
if n, err = strconv.ParseFloat(strings.TrimSpace(s), 64); err != nil {
return 0, fmt.Errorf("is not a number")
}
}
if n <= 0 || math.IsNaN(n) || math.IsInf(n, 0) {
return 0, fmt.Errorf("is not above zero")
}
return n, nil
}
+89 -2
View File
@@ -3,7 +3,9 @@ package buildstep
import (
"os"
"path/filepath"
"strings"
"testing"
"time"
)
// 1887/A7: the node's build capacity is the build module's setting, read at every pull: the control-node at 0 pulls
@@ -37,8 +39,10 @@ func TestNodeCapacity(t *testing.T) {
}
}
// ADR 0553 §4: a step's threads are the composer's share over the module's own capacity; without the word, none.
func TestNodeLimits(t *testing.T) {
// ADR 0553 §4: a step's threads are the composer's share over the module's own capacity; without the word, none. And
// issue 1806's follow-up: the capacity judged by is the configured one bounded by the build slice's cores over the cores
// a step needs (the three build nodes as they stand on 2026-10-26), and cores-per-step is read from the settings file.
func TestNodeLimitsAndTheCPUBound(t *testing.T) {
t.Setenv("MESH_BUILD_CPU_SHARE", "")
if l := NodeLimits(3); l.Threads != 0 || l.Slots != 3 {
t.Errorf("without the word: %+v", l)
@@ -47,4 +51,87 @@ func TestNodeLimits(t *testing.T) {
if l := NodeLimits(3); l.Percent != 75 || l.Slots != 3 || l.Threads < 1 {
t.Errorf("from the word: %+v", l)
}
for _, c := range []struct {
name string
configured, quota int
cores float64
effective, byCPU int
}{
{"shanks: 12 cores, configured 3", 3, 1200, 6, 2, 2},
{"ace: 18 cores, configured 2", 2, 1800, 6, 2, 3},
{"g14: 8 cores, configured 2", 2, 800, 6, 1, 1},
{"node on shanks: 12 cores at 2", 3, 1200, 2, 3, 6},
{"no quota: the configured one", 3, 0, 6, 3, 0},
{"configured 0 pulls nothing", 0, 1200, 6, 0, 0},
{"a quota under one step still builds one", 2, 100, 6, 1, 1},
{"no cores a step is one", 2, 150, 0, 1, 1},
} {
got := EffectiveCapacity(c.configured, c.quota, "test", c.cores)
if got.Effective != c.effective || got.ByCPU != c.byCPU || got.Why == "" {
t.Errorf("%s: %+v, want effective %d, by CPU %d", c.name, got, c.effective, c.byCPU)
}
}
file := filepath.Join(t.TempDir(), "settings.json")
var said []string
cores := NodeCoresPerStep(file, 6, func(f string, a ...any) { said = append(said, f) })
for _, c := range []struct {
file string
want float64
}{{`{"capacity":"shanks=2"}`, 6}, {`{"cores-per-step":4}`, 4}, {`{"cores-per-step":"3.5"}`, 3.5}, {`{"cores-per-step":0}`, 6}, {`{"cores-per-step":0}`, 6}} {
os.WriteFile(file, []byte(c.file), 0o600)
if got := cores(); got != c.want {
t.Errorf("cores-per-step from %s = %g, want %g", c.file, got, c.want)
}
}
if len(said) != 1 {
t.Errorf("a value not above zero is said once, said %d times", len(said))
}
// The operator, 2026-10-26 21:18: g14 builds only from 18:00 to 06:00, its local time. Outside its build hours the
// capacity judged by is 0, said with the window and when it opens; a node not named, or a bad value, builds at any hour.
cet := time.FixedZone("CET", 3600)
day := func(hh, mm int) time.Time { return time.Date(2026, 10, 26, hh, mm, 0, 0, cet) }
was := now
defer func() { now = was }()
for _, c := range []struct {
name, setting, node string
at time.Time
effective int
edge string // the window's next edge, HH:MM; "" for no window
bad bool
}{
{"g14 in the office", "g14=18:00-06:00", "g14", day(10, 0), 0, "18:00", false},
{"g14 in the evening", "g14=18:00-06:00", "g14", day(18, 0), 2, "06:00", false},
{"g14 past midnight", "g14=18:00-06:00", "g14", day(5, 59), 2, "06:00", false},
{"g14 at the close", "g14=18:00-06:00", "g14", day(6, 0), 0, "18:00", false},
{"a day window", "shanks=09:00-17:30,g14=18:00-06:00", "shanks", day(17, 29), 2, "17:30", false},
{"a node not named", "g14=18:00-06:00", "ace", day(10, 0), 2, "", false},
{"a bad time", "g14=18:00-6", "g14", day(10, 0), 2, "", true},
{"no span", "g14=18:00-18:00", "g14", day(10, 0), 2, "", true},
} {
w, err := ParseBuildHours(c.setting, c.node)
if (err != nil) != c.bad {
t.Errorf("%s: parse %v", c.name, err)
}
if edge := ""; w != nil {
edge = w.NextEdge(c.at).Format("15:04")
if edge != c.edge {
t.Errorf("%s: next edge %s, want %s", c.name, edge, c.edge)
}
} else if c.edge != "" {
t.Errorf("%s: no window, want one", c.name)
}
file := filepath.Join(t.TempDir(), "settings.json")
os.WriteFile(file, []byte(`{"build-hours":"`+c.setting+`"}`), 0o600)
now = func() time.Time { return c.at }
p := &Puller{Language: "go", Node: c.node, Capacity: func() int { return 2 }, BuildHours: NodeBuildHours(file, c.node, nil)}
got := p.CapacityDetail()
if got.Effective != c.effective || got.OutsideHours != (c.effective == 0) || (c.effective == 0 && !strings.Contains(got.Why, "outside its build hours 18:00-06:00 CET until 18:00")) {
t.Errorf("%s: %+v, want capacity %d", c.name, got, c.effective)
}
// A value that cannot be read builds at any hour, loudly: the record names the value and says why.
if c.bad && (!strings.Contains(got.BuildHoursFault, c.setting) || !strings.Contains(got.Why, "at any hour, because its build hours cannot be read")) {
t.Errorf("%s: a bad value is not said in the record: %+v", c.name, got)
}
}
}
+13
View File
@@ -15,6 +15,9 @@ type MergeResult struct {
Base, Head, Merge string
Conflicts []string
Changed []string
// Programs are the changed files that are compiled programs (ExecutableKind of their first bytes, read in the workspace
// at the merge); never nil for a clean merge, so a caller tells none from a module that reports none (issue 1577).
Programs []Program
}
// MergeInto makes, in the workspace dir that FetchCommit left at commit, the merge of commit into the trunk commit
@@ -98,6 +101,16 @@ func MergeInto(ctx context.Context, run forgekey.Runner, dir, remote, base, comm
return res, fmt.Errorf("the paths the merge changes cannot be read: %w", err)
}
res.Changed = strings.Fields(changed)
raw, err := git("diff", "--raw", "-z", "--no-renames", "--diff-filter=AM", base, res.Merge)
if err != nil {
return res, fmt.Errorf("the files the merge adds or modifies cannot be read: %w", err)
}
res.Programs = programsIn(dir, raw)
history, err := historyPrograms(git, base, commit, res.Programs)
if err != nil {
return res, fmt.Errorf("the commits of the branch cannot be read for compiled programs: %w", err)
}
res.Programs = append(res.Programs, history...)
return res, nil
}
+30 -3
View File
@@ -79,6 +79,30 @@ func (r *mergeRig) head(files map[string]string) string {
return h
}
// headThatAddsAndRemoves is head with an earlier commit on the branch that adds a program and a later one that removes it: the
// branch's tree is clean, its history is not. It answers the head and the earlier commit.
func (r *mergeRig) headThatAddsAndRemoves(files map[string]string) (head, added string) {
realGit(r.t, r.work, "checkout", "-q", "-B", "pr", r.fork)
writeFile(r.t, r.work, "gone/prog", "\x7fELF\x02\x01")
realGit(r.t, r.work, "add", ".")
realGit(r.t, r.work, "commit", "-q", "-m", "oops")
added = realGit(r.t, r.work, "rev-parse", "HEAD")
realGit(r.t, r.work, "rm", "-q", "gone/prog")
return r.head2(files), added
}
// head2 commits files on the current branch and pushes it as the pull request's head.
func (r *mergeRig) head2(files map[string]string) string {
for n, c := range files {
writeFile(r.t, r.work, n, c)
}
realGit(r.t, r.work, "add", ".")
realGit(r.t, r.work, "commit", "-q", "-m", "head")
h := realGit(r.t, r.work, "rev-parse", "HEAD")
realGit(r.t, r.work, "push", "-q", r.bare, "HEAD:refs/pull/1/head")
return h
}
// advance moves the trunk past the fork point with a commit changing files, pushes it and answers its sha.
func (r *mergeRig) advance(files map[string]string) string {
realGit(r.t, r.work, "checkout", "-q", "-B", "main", r.fork)
@@ -102,7 +126,7 @@ func (r *mergeRig) fetch(head string) {
// holds both sides' changes.
func TestMergeIntoACleanMergeAnswersTheMergeCommitAndTheChangedFiles(t *testing.T) {
r := newMergeRig(t)
head := r.head(map[string]string{"b.txt": "head\n"})
head, added := r.headThatAddsAndRemoves(map[string]string{"b.txt": "head\n", "prog": "\x7fELF\x02\x01"})
base := r.advance(map[string]string{"a.txt": "one\ntwo\nthree\nfour\n"})
r.fetch(head)
res, err := MergeInto(context.Background(), forgekey.ExecRunner, r.ws, r.remote, base, head, AllowFileRemote())
@@ -112,8 +136,11 @@ func TestMergeIntoACleanMergeAnswersTheMergeCommitAndTheChangedFiles(t *testing.
if res.Base != base || res.Head != head || len(res.Merge) != 40 || res.Merge == base || res.Merge == head || len(res.Conflicts) != 0 {
t.Fatalf("result = %+v", res)
}
if !reflect.DeepEqual(res.Changed, []string{"b.txt"}) {
t.Errorf("changed = %v, want only what the head adds to the trunk: b.txt", res.Changed)
if !reflect.DeepEqual(res.Changed, []string{"b.txt", "prog"}) {
t.Errorf("changed = %v, want only what the head adds to the trunk: b.txt, prog", res.Changed)
}
if !reflect.DeepEqual(res.Programs, []Program{{"prog", "ELF", ""}, {"gone/prog", "ELF", added}}) { // issue 1577
t.Errorf("programs = %v, want the ELF the head adds and the one an earlier commit added and a later one removed", res.Programs)
}
if at := realGit(t, r.ws, "rev-parse", "HEAD"); at != res.Merge {
t.Errorf("the workspace is at %s, not at the merge %s", at, res.Merge)
+67
View File
@@ -0,0 +1,67 @@
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
}
+115
View File
@@ -0,0 +1,115 @@
package buildstep
import (
"bytes"
"io"
"os"
"path/filepath"
"strings"
)
// Program is a changed file whose first bytes are an executable binary: a compiled program, which is build output and not
// source (novox/hq issue 1577: `go build` in a module's folder writes one beside go.mod, and `git add -A` takes it).
type Program struct {
Path string `json:"path"`
Kind string `json:"kind"` // ELF, Mach-O or PE
// Commit is set for a program a commit of the branch added and a later one removed (or replaced): the merge's tree is clean,
// but the merge commit carries the blob into the trunk's history for good. Empty for a program the merge's tree holds.
Commit string `json:"commit,omitempty"`
}
// ExecutableKind says what executable format the first bytes of a file are, or "" when they are none. A PE needs its header
// ("PE\0\0" at the offset in bytes 0x3c) so a text starting with "MZ" is not one; a fat Mach-O (0xCAFEBABE) is told from a
// Java class by its architecture count (bytes 4 to 7, fewer than 0x20; a class has its version there).
func ExecutableKind(head []byte) string {
switch {
case bytes.HasPrefix(head, []byte("\x7fELF")):
return "ELF"
case bytes.HasPrefix(head, []byte("MZ")):
if len(head) >= 0x40 {
off := int(head[0x3c]) | int(head[0x3d])<<8 | int(head[0x3e])<<16 | int(head[0x3f])<<24
if off >= 0 && off+4 <= len(head) && bytes.Equal(head[off:off+4], []byte("PE\x00\x00")) {
return "PE"
}
}
case bytes.HasPrefix(head, []byte{0xcf, 0xfa, 0xed, 0xfe}), bytes.HasPrefix(head, []byte{0xce, 0xfa, 0xed, 0xfe}),
bytes.HasPrefix(head, []byte{0xfe, 0xed, 0xfa, 0xcf}), bytes.HasPrefix(head, []byte{0xfe, 0xed, 0xfa, 0xce}),
bytes.HasPrefix(head, []byte{0xca, 0xfe, 0xba, 0xbe}) && len(head) >= 8 && head[4] == 0 && head[5] == 0 && head[6] == 0 && head[7] < 0x20:
return "Mach-O"
}
return ""
}
// programHeadSize is how many bytes of a file are read to tell its format: enough for a PE header at its usual offsets.
const programHeadSize = 0x100
// historyPrograms names the compiled programs that a commit of base..head added or modified and the merge's tree no longer
// holds at that path (those it holds are in final already): each with the commit that added it. git is the merge's git runner.
func historyPrograms(git func(args ...string) (string, error), base, head string, final []Program) ([]Program, error) {
revs, err := git("rev-list", "--no-merges", "--reverse", base+".."+head)
if err != nil {
return nil, err
}
held := map[string]bool{}
for _, p := range final {
held[p.Path] = true
}
out, seen := []Program{}, map[string]bool{}
for _, c := range strings.Fields(revs) {
raw, err := git("diff-tree", "-r", "-z", "--no-commit-id", "--no-renames", "--root", "--diff-filter=AM", c)
if err != nil {
return nil, err
}
f := strings.Split(raw, "\x00")
for i := 0; i+1 < len(f); i += 2 {
meta := strings.Fields(strings.TrimPrefix(f[i], ":"))
if len(meta) < 4 || (meta[1] != "100644" && meta[1] != "100755") || held[f[i+1]] || seen[meta[3]] {
continue
}
seen[meta[3]] = true
blob, err := git("cat-file", "blob", meta[3])
if err != nil {
return nil, err
}
head := []byte(blob)
if len(head) > programHeadSize {
head = head[:programHeadSize]
}
if k := ExecutableKind(head); k != "" {
out = append(out, Program{Path: f[i+1], Kind: k, Commit: c})
}
}
}
return out, nil
}
// programsIn reads the output of `git diff --raw -z --no-renames --diff-filter=AM` (":<old mode> <new mode> <old sha> <new
// sha> <status>\0<path>\0" per file) and names each added or modified regular file of the tree dir that is a compiled
// program, from its first bytes read locally. A symlink (mode 120000) and a submodule (160000) are not read. It never
// answers nil, so a caller can tell no programs from none reported.
func programsIn(dir, raw string) []Program {
out := []Program{}
f := strings.Split(raw, "\x00")
for i := 0; i+1 < len(f); i += 2 {
meta := strings.Fields(strings.TrimPrefix(f[i], ":"))
if len(meta) < 2 || (meta[1] != "100644" && meta[1] != "100755") {
continue
}
p := f[i+1]
full := filepath.Join(dir, filepath.FromSlash(p))
if st, err := os.Lstat(full); err != nil || !st.Mode().IsRegular() {
continue
}
fh, err := os.Open(full)
if err != nil {
continue
}
head := make([]byte, programHeadSize)
n, _ := io.ReadFull(fh, head)
fh.Close()
if k := ExecutableKind(head[:n]); k != "" {
out = append(out, Program{Path: p, Kind: k})
}
}
return out
}
+72
View File
@@ -0,0 +1,72 @@
package buildstep
import (
"os"
"path/filepath"
"reflect"
"testing"
)
func pe(offset int, magic string) []byte {
b := make([]byte, 0x100)
copy(b, "MZ")
b[0x3c] = byte(offset)
copy(b[offset:], magic)
return b
}
func TestExecutableKindTellsProgramsFromText(t *testing.T) { // issue 1577
for _, c := range []struct {
name string
head []byte
want string
}{
{"elf", []byte("\x7fELF\x02\x01\x01"), "ELF"},
{"mach-o 64", []byte{0xcf, 0xfa, 0xed, 0xfe, 7, 0, 0, 1}, "Mach-O"},
{"mach-o 32 big", []byte{0xfe, 0xed, 0xfa, 0xce, 0, 0, 0, 7}, "Mach-O"},
{"fat mach-o", []byte{0xca, 0xfe, 0xba, 0xbe, 0, 0, 0, 2}, "Mach-O"},
{"java class", []byte{0xca, 0xfe, 0xba, 0xbe, 0, 0, 0, 0x34}, ""},
{"pe", pe(0x80, "PE\x00\x00"), "PE"},
{"mz text", []byte("MZ is a text file starting with two letters"), ""},
{"mz without a pe header", pe(0x80, "XX\x00\x00"), ""},
{"text", []byte("package main"), ""},
{"empty", nil, ""},
} {
if got := ExecutableKind(c.head); got != c.want {
t.Errorf("%s: kind = %q, want %q", c.name, got, c.want)
}
}
}
func TestProgramsInNamesTheAddedAndModifiedRegularFilesThatArePrograms(t *testing.T) { // issue 1577
dir := t.TempDir()
write := func(name string, b []byte) {
if err := os.MkdirAll(filepath.Dir(filepath.Join(dir, name)), 0o755); err != nil {
t.Fatal(err)
}
if err := os.WriteFile(filepath.Join(dir, name), b, 0o755); err != nil {
t.Fatal(err)
}
}
write("m/prog", []byte("\x7fELF\x02\x01"))
write("m/notes", []byte("plain text"))
write("m/mac", []byte{0xcf, 0xfa, 0xed, 0xfe, 7, 0, 0, 1})
if err := os.Symlink("prog", filepath.Join(dir, "m/link")); err != nil {
t.Fatal(err)
}
line := func(mode, path string) string { return ":000000 " + mode + " 0000 1111 A\x00" + path + "\x00" }
for _, c := range []struct {
name, raw string
want []Program
}{
{"none", "", []Program{}},
{"a program and a text", line("100755", "m/prog") + line("100644", "m/notes"), []Program{{"m/prog", "ELF", ""}}},
{"two programs", line("100644", "m/prog") + line("100755", "m/mac"), []Program{{"m/prog", "ELF", ""}, {"m/mac", "Mach-O", ""}}},
{"a symlink and a submodule are not read", line("120000", "m/link") + line("160000", "m/sub"), []Program{}},
{"a path the tree lacks", line("100644", "m/gone"), []Program{}},
} {
if got := programsIn(dir, c.raw); !reflect.DeepEqual(got, c.want) {
t.Errorf("%s: programs = %v, want %v", c.name, got, c.want)
}
}
}
+201 -13
View File
@@ -77,6 +77,15 @@ type Puller struct {
// for this node, and wakes the puller on each change (watchCapacity): WatchConfigServed in a module, a test's own
// double. Until it hands a value the puller judges by Capacity.
WatchConfig func(key string, handle func(stdio.ProvisionChange) error) error
// CoresPerStep is what one ask of this module needs, read at every judgement (NodeCoresPerStep); with it the capacity
// judged by is bounded by the node's build slice: min(the configured capacity, floor(the quota's cores / CoresPerStep))
// (EffectiveCapacity). Nil bounds nothing by the CPU. Quota reads the quota (Quota when nil; a test's own).
CoresPerStep func() float64
Quota func() (percent int, from string)
// BuildHours is the node's build window, read at every judgement (NodeBuildHours); with WatchConfig set the value
// mesh-config resolves for `build-hours` takes its place once handed. Outside the window the capacity judged by is 0:
// steps running finish, no new one is taken; the puller wakes at the window's next edge. Nil builds at any hour.
BuildHours func() (*BuildWindow, string)
// Has says whether this node has a need an ask names (delivery.AskNeeds, issue 890). An ask needing what it lacks is
// declined: its step ends refused as declined (delivery.StepDeclined), the message is dropped for its asker to put
// again, and the worker it came from is not looked at again until the next pull's wait, so a node lacking the need
@@ -92,6 +101,16 @@ type Puller struct {
// watched is the capacity the config watch handed last, used in place of Capacity once hasWatched.
watched int
hasWatched bool
// hours is the build window the config watch handed last, and hoursFault why its value cannot be read; used in place
// of BuildHours once hasHours.
hours *BuildWindow
hoursFault string
hasHours bool
// judged is the capacity said last (log and capacity.<node>), so a change is said once.
judged delivery.Capacity
hasJudged bool
capWritten bool
capFault string
// declined are the kinds whose worker handed an ask this node declined, left out of the pulls without a wait until
// the next pull's wait ends.
declined map[string]bool
@@ -216,23 +235,128 @@ func (p *Puller) room() bool {
// roomNow is room, with the capacity and the asks in hand it was judged on.
func (p *Puller) roomNow() (ok bool, capacity, inHand int) {
p.mu.Lock()
watched, has := p.watched, p.hasWatched
p.mu.Unlock()
if has {
capacity = watched
} else if p.Capacity != nil {
capacity = p.Capacity()
}
capacity = p.CapacityDetail().Effective
p.mu.Lock()
defer p.mu.Unlock()
return !p.paused && p.inHand < capacity, capacity, p.inHand
}
// CapacityNow is the capacity the puller judges by: the config watch's last value, else Capacity.
// CapacityNow is the capacity the puller judges by (CapacityDetail's Effective).
func (p *Puller) CapacityNow() int {
_, capacity, _ := p.roomNow()
return capacity
return p.CapacityDetail().Effective
}
// CapacityDetail is the capacity the puller judges by now and why: the configured one (the config watch's last value,
// else Capacity), bounded by the node's build slice when CoresPerStep is set (EffectiveCapacity), its quota read now. A
// change from what it said last is logged and written to `capacity.<node>` of the pauses state (PausePut), so the
// delivery view says it too.
func (p *Puller) CapacityDetail() delivery.Capacity {
p.mu.Lock()
watched, has := p.watched, p.hasWatched
p.mu.Unlock()
configured, from := 0, "the settings file"
switch {
case has:
configured, from = watched, "mesh-config, as the config watch handed it"
case p.Capacity != nil:
configured = p.Capacity()
}
var c delivery.Capacity
if p.CoresPerStep == nil {
c = delivery.Capacity{Configured: configured, Effective: configured, QuotaFrom: "not judged: the module names no cores a step needs",
Why: fmt.Sprintf("its configured capacity %d", configured)}
} else {
quota := p.Quota
if quota == nil {
quota = Quota
}
percent, qfrom := quota()
c = EffectiveCapacity(configured, percent, qfrom, p.CoresPerStep())
}
c.ConfiguredFrom = from
w, fault := p.window()
if fault != "" {
c.BuildHoursFault = fault
c.Why = fmt.Sprintf("%s; at any hour, because its build hours cannot be read: %s", c.Why, fault)
}
if w != nil {
at := now()
zone, _ := at.Zone()
c.BuildHours = w.String() + " " + zone
if !w.Open(at) && c.Effective > 0 {
c.OutsideHours = true
c.Why = fmt.Sprintf("0, outside its build hours %s until %s (otherwise %s)", c.BuildHours, w.NextEdge(at).Format("15:04"), c.Why)
c.Effective = 0
} else if c.Effective > 0 {
c.Why = fmt.Sprintf("%s; within its build hours %s until %s", c.Why, c.BuildHours, w.NextEdge(at).Format("15:04"))
}
}
p.sayCapacity(c)
return c
}
// window is the node's build window now (the config watch's last value, else BuildHours, else none), and why its value
// cannot be read when it cannot.
func (p *Puller) window() (*BuildWindow, string) {
p.mu.Lock()
w, fault, has := p.hours, p.hoursFault, p.hasHours
p.mu.Unlock()
if has {
return w, fault
}
if p.BuildHours != nil {
return p.BuildHours()
}
return nil, ""
}
// untilEdge is how long until the build window's next edge, and false when the node has no window.
func (p *Puller) untilEdge() (time.Duration, bool) {
w, _ := p.window()
if w == nil {
return 0, false
}
at := now()
return w.NextEdge(at).Sub(at), true
}
// sayCapacity logs c when it differs from what was said last, and writes it to the node's capacity key until a write
// lands (a failed write is tried again at the next judgement, its error said once).
func (p *Puller) sayCapacity(c delivery.Capacity) {
p.mu.Lock()
same := p.hasJudged && p.judged.Effective == c.Effective && p.judged.Configured == c.Configured && p.judged.ByCPU == c.ByCPU &&
p.judged.CoresPerStep == c.CoresPerStep && p.judged.QuotaCores == c.QuotaCores && p.judged.OutsideHours == c.OutsideHours &&
p.judged.BuildHours == c.BuildHours && p.judged.BuildHoursFault == c.BuildHoursFault
if !same {
p.judged, p.hasJudged, p.capWritten = c, true, false
}
write := !p.capWritten && p.PausePut != nil
p.mu.Unlock()
if !same {
p.logf("the %s build capacity of %s is %s (configured from %s)", p.Language, p.Node, c.Why, c.ConfiguredFrom)
}
if !write {
return
}
key, err := delivery.CapacityKey(p.Node)
if err != nil {
return
}
c.At = now()
raw, err := delivery.EncodeCapacity(c)
if err == nil {
_, err = p.PausePut(key, json.RawMessage(raw))
}
p.mu.Lock()
defer p.mu.Unlock()
if err == nil {
p.capWritten, p.capFault = true, ""
return
}
if err.Error() != p.capFault {
p.capFault = err.Error()
p.logf("the %s build capacity of %s could not be written to %s (tried again at the next judgement): %v", p.Language, p.Node, key, err)
}
}
// Pull takes and runs asks until ctx ends. An ask still running then is neither ended nor acknowledged: the stream
@@ -248,6 +372,7 @@ func (p *Puller) Pull(ctx context.Context) error {
p.logf("the pause of %s cannot be read: %v; pulling as not paused", p.Node, err)
}
go p.watchCapacity() // WatchConfigServed waits for Serve, which a module starts after its pull
go p.watchHours()
wait := p.PullWait
if wait <= 0 {
wait = DefaultPullWait
@@ -260,16 +385,29 @@ func (p *Puller) Pull(ctx context.Context) error {
// Parked until something wakes it: a capacity change (the config watch), a resume, or a step's end. Every
// park logs once, so a puller that sits idle says why (issue 1887).
p.logf("the %s puller on %s parks: %d of capacity %d in hand, paused %v", p.Language, p.Node, inHand, capacity, p.Paused())
var edge <-chan time.Time // the build window's next edge, when the node has one: one timer, not a poll
stop := func() bool { return false }
if d, ok := p.untilEdge(); ok {
t := time.NewTimer(d + time.Second)
edge, stop = t.C, t.Stop
}
select {
case <-ctx.Done():
stop()
return ctx.Err()
case <-p.wake:
case <-edge:
}
stop()
_, capacity, inHand = p.roomNow()
p.logf("the %s puller on %s wakes: %d of capacity %d in hand, paused %v", p.Language, p.Node, inHand, capacity, p.Paused())
continue
}
msg, err := p.next(ctx, wait)
pullWait := wait
if d, ok := p.untilEdge(); ok && d < pullWait {
pullWait = d // a pull does not wait past the window's close
}
msg, err := p.next(ctx, pullWait)
if err != nil {
if ctx.Err() != nil {
return ctx.Err()
@@ -395,6 +533,7 @@ func (p *Puller) take(ctx context.Context, msg *Pulled) {
}
p.running[ask.ID] = RunningAsk{ID: ask.ID, Kind: ask.Kind, Module: module, Started: step.Started}
p.mu.Unlock()
p.logf("the %s ask %s starts on %s: %s", ask.Kind, ask.ID, p.Node, step.What)
go p.run(ctx, msg, ask, key, step)
}
@@ -425,7 +564,11 @@ func (p *Puller) run(ctx context.Context, msg *Pulled, ask delivery.StepAsk, key
}
}
}()
result, err := protect(ctx, func(ctx context.Context) (any, error) { return p.Run(ctx, ask) })
ph := &phases{put: p.Put, key: key, step: step, logf: p.Logf, prefix: fmt.Sprintf("the %s ask %s on %s", ask.Kind, ask.ID, p.Node)}
result, err := protect(ctx, func(ctx context.Context) (any, error) {
return p.Run(context.WithValue(ctx, phaseKey{}, ph), ask)
})
step = ph.end() // its end keeps the phase it ended in
close(done)
if ctx.Err() != nil {
// The process stops: the ask is neither ended nor acknowledged here, so the stream gives it to a node that runs.
@@ -438,6 +581,15 @@ func (p *Puller) run(ctx context.Context, msg *Pulled, ask delivery.StepAsk, key
_ = msg.Nak()
return
}
outcome := "passed"
if err != nil {
outcome = "did not pass: " + err.Error()
if len(outcome) > 300 {
outcome = outcome[:300] + "..."
}
}
p.logf("the %s ask %s ended on %s after %s, in its phase %q; %s", ask.Kind, ask.ID, p.Node,
now().Sub(step.Started).Round(time.Second), step.Phase, outcome)
if aerr := msg.Ack(); aerr != nil {
p.logf("the ask %s ended on %s and its acknowledgement failed: %v", ask.ID, p.Node, aerr)
}
@@ -474,6 +626,42 @@ func (p *Puller) watchCapacity() {
}
}
// watchHours follows the module's config value `build-hours` when WatchConfig and BuildHours are set, as watchCapacity
// follows `capacity`: each value handed is kept, used in place of BuildHours, and wakes the puller. A value that cannot
// be read is no window, said; a watch that cannot start leaves the puller on BuildHours, said once.
func (p *Puller) watchHours() {
if p.WatchConfig == nil || p.BuildHours == nil {
return
}
err := p.WatchConfig(BuildHoursSetting, func(c stdio.ProvisionChange) error {
setting := ""
var w *BuildWindow
var why string
switch {
case c.Refused != "":
why = "its build hours are no longer answered: " + c.Refused
case len(c.Value) > 0 && string(c.Value) != "null" && json.Unmarshal(c.Value, &setting) != nil:
why = fmt.Sprintf("the setting %s %s is not a string", BuildHoursSetting, c.Value)
default:
var err error
if w, err = ParseBuildHours(setting, p.Node); err != nil {
why = err.Error()
}
}
p.mu.Lock()
p.hours, p.hoursFault, p.hasHours = w, why, true
p.mu.Unlock()
if why != "" {
p.logf("the %s puller on %s builds at any hour: %s", p.Language, p.Node, why)
}
p.Wake()
return nil
})
if err != nil {
p.logf("the %s build hours of %s cannot be followed on the bus, so they are read from the settings file: %v", p.Language, p.Node, err)
}
}
// WatchConfigServed is stdio.WatchConfig once Serve has the runtime's channel (stdio.Served): a build module starts its
// pull before it serves, and a watch asked before that would be refused.
func WatchConfigServed(key string, handle func(stdio.ProvisionChange) error) error {
+58 -3
View File
@@ -104,20 +104,38 @@ func TestPullTakesDeliveryBeforeCheck(t *testing.T) {
}
// 1887/A2a, 1887/AC-pull: given capacity 2 and three asks waiting, two run and the third waits, and it starts when
// one ends, with no timer.
// one ends, with no timer. Issue 1806's follow-up: the 2 is the CPU's, a configured 3 bounded by a 12-core slice at 6
// cores a step, said once in capacity.<node>; and issue 1890: a run's phase is written to its step key and logged with
// its ask id, and its end keeps the last phase.
func TestPullHoldsAtMostItsCapacity(t *testing.T) {
src, st := NewFakeAsks(), newSteps()
asks := []delivery.StepAsk{}
for i := 0; i < 3; i++ {
_ = src.PublishAsk(testAsk(delivery.AskDelivery))
a := testAsk(delivery.AskDelivery)
asks = append(asks, a)
_ = src.PublishAsk(a)
}
release := make(chan struct{})
var mu sync.Mutex
started := 0
p := &Puller{Source: src, Language: delivery.AskGo, Node: "shanks", Put: st.put, Capacity: func() int { return 2 },
var said []string
capacities := map[string][]byte{}
quota := 1200
p := &Puller{Source: src, Language: delivery.AskGo, Node: "shanks", Put: st.put, Capacity: func() int { return 3 },
CoresPerStep: func() float64 { return 6 },
Quota: func() (int, string) { mu.Lock(); defer mu.Unlock(); return quota, "a test's slice" },
PausePut: func(key string, v any) (uint64, error) {
mu.Lock()
capacities[key] = v.(json.RawMessage)
mu.Unlock()
return 1, nil
},
Logf: func(f string, a ...any) { mu.Lock(); said = append(said, fmt.Sprintf(f, a...)); mu.Unlock() },
PullWait: 10 * time.Millisecond, Run: func(ctx context.Context, a delivery.StepAsk) (any, error) {
mu.Lock()
started++
mu.Unlock()
Phase(ctx, "unit: tests of 3 packages")
<-release
return nil, nil
}}
@@ -129,12 +147,49 @@ func TestPullHoldsAtMostItsCapacity(t *testing.T) {
if p.InHand() != 2 || src.Waiting(delivery.AskDelivery) != 1 {
t.Fatalf("in hand %d, waiting %d; want 2 and 1", p.InHand(), src.Waiting(delivery.AskDelivery))
}
c := p.CapacityDetail()
if c.Effective != 2 || c.Configured != 3 || c.ByCPU != 2 || c.QuotaCores != 12 || !strings.Contains(c.Why, "bounded by the CPU") {
t.Fatalf("the capacity judged by = %+v, want 2 of a configured 3 bounded by 12 cores / 6", c)
}
mu.Lock()
raw := capacities["capacity.shanks"]
capLines := 0
for _, l := range said {
if strings.Contains(l, "build capacity of shanks is 2, bounded by the CPU") {
capLines++
}
}
mu.Unlock()
if got, err := delivery.DecodeCapacity(raw); err != nil || got.Effective != 2 || capLines != 1 {
t.Fatalf("capacity.shanks = %s (%v), said %d times; want effective 2, said once", raw, err, capLines)
}
waitFor(t, "a phase on the step key", func() bool {
key, _ := delivery.AskStepKey("shanks", p.Running()[0].ID)
s, _ := st.get(key)
return s.State == delivery.StepRunning && s.Phase == "unit: tests of 3 packages" && !s.PhaseAt.IsZero()
})
release <- struct{}{}
waitFor(t, "the third taken", func() bool { return src.Waiting(delivery.AskDelivery) == 0 })
// Taken from the source comes a moment before its run starts: wait for the start, not only the take.
waitFor(t, "the third started", func() bool { mu.Lock(); defer mu.Unlock(); return started == 3 })
close(release)
waitFor(t, "all acked", func() bool { a, _, _ := src.Fates(); return len(a) == 3 })
for _, a := range asks {
key, _ := delivery.AskStepKey("shanks", a.ID)
if s, _ := st.get(key); s.State != delivery.StepPassed || s.Phase != "unit: tests of 3 packages" {
t.Errorf("the end of %s = %+v, want passed in its last phase", a.ID, s)
}
}
mu.Lock()
logged := strings.Join(said, "\n")
quota = 600 // the cap lowered: one step at 6 cores, followed at the next judgement
mu.Unlock()
if !strings.Contains(logged, "the delivery ask "+asks[0].ID+" on shanks: unit: tests of 3 packages") {
t.Errorf("no journal line names the ask's phase: %s", logged)
}
if c := p.CapacityDetail(); c.Effective != 1 {
t.Errorf("after the quota fell to 6 cores, capacity %+v, want 1", c)
}
}
// An ask that is not one, or not this seat's, is dropped and runs nowhere. And issue 890/AC1: a check ask needing
+5 -2
View File
@@ -14,10 +14,13 @@ func SeatTools(seat, what string, p *Puller) []stdio.Tool {
Run: func(map[string]any) (any, error) { return run() }}
}
current := func() (any, error) {
return map[string]any{"node": p.Node, "capacity": p.CapacityNow(), "running": p.Running(), "paused": p.Paused()}, nil
c := p.CapacityDetail()
return map[string]any{"node": p.Node, "capacity": c.Effective, "capacity_why": c.Why, "capacity_detail": c,
"running": p.Running(), "paused": p.Paused()}, nil
}
return []stdio.Tool{
verb("current", "The "+what+" asks this node runs now (each one's id, kind and when it started), its build capacity, "+
verb("current", "The "+what+" asks this node runs now (each one's id, kind and when it started), its build capacity (the configured one, "+
"bounded by the cores its node's build slice gives over the cores a step needs) and why, "+
"and whether it is paused.", current),
verb("pause", "Take no new "+what+" ask on this node until resumed; an ask running here ends in its place. Kept "+
"across a restart (the node's pause key).", func() (any, error) {
+17
View File
@@ -86,6 +86,9 @@ func (k Asker) Wait(ctx context.Context, a delivery.StepAsk) (delivery.Step, err
if !ok {
return delivery.Step{}, errors.New("the watch of the ask's step ended")
}
if seen := stepSeenOf(ctx); seen != nil {
seen(step)
}
if step.State != delivery.StepRunning {
return step, nil
}
@@ -108,6 +111,20 @@ func (k Asker) Wait(ctx context.Context, a delivery.StepAsk) (delivery.Step, err
}
}
type stepSeenKey struct{}
// WithStepSeen is ctx carrying seen, which Asker.Wait hands every value the ask's step key takes while it waits: the node
// that took it, each phase it says while running (delivery.Step.Phase), and its end. A try names its asks' progress so
// (issue 1890). seen runs in the waiting goroutine and must not block.
func WithStepSeen(ctx context.Context, seen func(delivery.Step)) context.Context {
return context.WithValue(ctx, stepSeenKey{}, seen)
}
func stepSeenOf(ctx context.Context) func(delivery.Step) {
seen, _ := ctx.Value(stepSeenKey{}).(func(delivery.Step))
return seen
}
// Ask puts a and waits for its step; a step that did not pass is an error naming its node, its state and its reason.
func (k Asker) Ask(ctx context.Context, a delivery.StepAsk) (delivery.Step, error) {
if err := k.Put(ctx, a); err != nil {
+4 -3
View File
@@ -26,9 +26,10 @@ func TestAnAskerPutsOnceAndReadsTheStep(t *testing.T) {
return delivery.Step{ID: a.ID, Node: "ace", State: delivery.StepPassed, Result: []byte(`{"artifacts":[]}`)}
}}
k := Asker{Publisher: f, Waiter: f}
step, err := k.Ask(context.Background(), goAsk("mvl0frv18d1957"))
if err != nil || step.Node != "ace" || len(f.Asks()) != 1 {
t.Fatalf("step %+v, %v, asks %d", step, err, len(f.Asks()))
var seen []delivery.Step // issue 1890: a waiter that asks to see each value of the step sees it
step, err := k.Ask(WithStepSeen(context.Background(), func(s delivery.Step) { seen = append(seen, s) }), goAsk("mvl0frv18d1957"))
if err != nil || step.Node != "ace" || len(f.Asks()) != 1 || len(seen) != 1 || seen[0].Node != "ace" {
t.Fatalf("step %+v, %v, asks %d, seen %+v", step, err, len(f.Asks()), seen)
}
f.Answer = func(a delivery.StepAsk) delivery.Step {
return delivery.Step{ID: a.ID, Node: "shanks", State: delivery.StepRefused, Reason: "go 1.27 installed"}
+57
View File
@@ -693,3 +693,60 @@ func DecodePause(data []byte) (Pause, error) {
}
return p, nil
}
// CapacityKey is the key, in a build module's PausesState, where the module on node says the capacity it judges by
// now and why: `capacity.<node>` (issue 1806's follow-up). The pauses state has no age, so the value stands until the
// module writes another; a reader such as the delivery view's build_usage says it in place of the configured capacity.
func CapacityKey(node string) (string, error) {
if err := checkName("node", node); err != nil {
return "", err
}
return "capacity." + node, nil
}
// Capacity is a build seat's capacity on one node as its holder judges by it: the configured capacity (the module's
// setting `capacity`), bounded by the cores its node's build slice gives (its CPU quota, read live) over the cores one
// step needs (the module's setting `cores-per-step`). Effective is min(Configured, ByCPU) when a quota is known, else
// Configured; one step at least while Configured is above zero, so a node whose quota is under one step's need still
// builds, slowly.
type Capacity struct {
V int `json:"v"`
Effective int `json:"effective"`
Configured int `json:"configured"`
// ConfiguredFrom says where Configured was read: mesh-config's value as the config watch handed it, or the settings
// file.
ConfiguredFrom string `json:"configured-from,omitempty"`
// QuotaCores is the build slice's CPU quota in processors (cpu.max: 1200% is 12), 0 when none is known; QuotaFrom
// says where it was read (the cgroup, or the share the composer gave) or why none is known.
QuotaCores float64 `json:"quota-cores,omitempty"`
QuotaFrom string `json:"quota-from"`
// CoresPerStep is what one step of this module needs; ByCPU is floor(QuotaCores / CoresPerStep), at least 1, and 0
// when no quota is known.
CoresPerStep float64 `json:"cores-per-step"`
ByCPU int `json:"by-cpu,omitempty"`
// BuildHours is the node's build window (HH:MM-HH:MM, its local time and zone), absent when it builds at any hour;
// OutsideHours says the time judged fell outside it, so Effective is 0 and the next step waits for the window to open.
BuildHours string `json:"build-hours,omitempty"`
OutsideHours bool `json:"outside-hours,omitempty"`
// BuildHoursFault is why the node's build-hours value cannot be read, naming the value: it then builds at any hour.
BuildHoursFault string `json:"build-hours-fault,omitempty"`
// Why is the capacity in words: what bounds it.
Why string `json:"why"`
At time.Time `json:"at,omitzero"`
}
// EncodeCapacity is c as JSON at Version.
func EncodeCapacity(c Capacity) ([]byte, error) {
c.V = Version
c.At = c.At.UTC()
return json.Marshal(c)
}
// DecodeCapacity reads a capacity value; it refuses an unknown version (ErrUnknownVersion).
func DecodeCapacity(data []byte) (Capacity, error) {
var c Capacity
if err := decode(data, &c); err != nil {
return Capacity{}, err
}
return c, nil
}
+5
View File
@@ -73,6 +73,11 @@ type Step struct {
// What says what the step is, for a reader of the steps state: an ask's kind, repository, head and levels or
// module (issue 1890). Empty for a step a tool started by hand.
What string `json:"what,omitempty"`
// Phase is the part of its job a running ask's step is in (fetch, a level's digest, lint or tests), and PhaseAt
// when it entered it: the step key is written again at each phase, so a slow step says where it stands (issue 1890).
// A reader waiting for the end treats such a write as any running one.
Phase string `json:"phase,omitempty"`
PhaseAt time.Time `json:"phase-at,omitzero"`
}
func (s Step) check() error {