From 22bc4dc8ef4ffd4eea6f705753bd8f010804237f Mon Sep 17 00:00:00 2001 From: jochen Date: Mon, 26 Oct 2026 18:59:35 +0100 Subject: [PATCH 1/4] MergeInto names the changed files that are compiled programs, read in the workspace, so a check can refuse build output (issue 1577) --- go/buildstep/merge.go | 8 +++ go/buildstep/merge_integration_test.go | 9 ++-- go/buildstep/programs.go | 72 ++++++++++++++++++++++++++ go/buildstep/programs_test.go | 72 ++++++++++++++++++++++++++ 4 files changed, 158 insertions(+), 3 deletions(-) create mode 100644 go/buildstep/programs.go create mode 100644 go/buildstep/programs_test.go diff --git a/go/buildstep/merge.go b/go/buildstep/merge.go index 907009e..589c50e 100644 --- a/go/buildstep/merge.go +++ b/go/buildstep/merge.go @@ -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,11 @@ 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) return res, nil } diff --git a/go/buildstep/merge_integration_test.go b/go/buildstep/merge_integration_test.go index 88b6d11..ce8518c 100644 --- a/go/buildstep/merge_integration_test.go +++ b/go/buildstep/merge_integration_test.go @@ -102,7 +102,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 := r.head(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 +112,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"}}) { // issue 1577 + t.Errorf("programs = %v, want the ELF the head adds", 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) diff --git a/go/buildstep/programs.go b/go/buildstep/programs.go new file mode 100644 index 0000000..8d87dce --- /dev/null +++ b/go/buildstep/programs.go @@ -0,0 +1,72 @@ +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 +} + +// 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 + +// programsIn reads the output of `git diff --raw -z --no-renames --diff-filter=AM` (": \0\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 +} diff --git a/go/buildstep/programs_test.go b/go/buildstep/programs_test.go new file mode 100644 index 0000000..7cc26bc --- /dev/null +++ b/go/buildstep/programs_test.go @@ -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) + } + } +} From fe11db9293973f098ef48eebe350a684574d2b39 Mon Sep 17 00:00:00 2001 From: jochen Date: Mon, 26 Oct 2026 19:47:17 +0100 Subject: [PATCH 2/4] MergeInto also names programs a commit of the branch added and a later one removed: the merge commit would keep the blob in history (issue 1577) --- go/buildstep/merge.go | 5 +++ go/buildstep/merge_integration_test.go | 30 ++++++++++++++++-- go/buildstep/programs.go | 43 ++++++++++++++++++++++++++ go/buildstep/programs_test.go | 4 +-- 4 files changed, 77 insertions(+), 5 deletions(-) diff --git a/go/buildstep/merge.go b/go/buildstep/merge.go index 589c50e..86384b6 100644 --- a/go/buildstep/merge.go +++ b/go/buildstep/merge.go @@ -106,6 +106,11 @@ func MergeInto(ctx context.Context, run forgekey.Runner, dir, remote, base, comm 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 } diff --git a/go/buildstep/merge_integration_test.go b/go/buildstep/merge_integration_test.go index ce8518c..f83ece8 100644 --- a/go/buildstep/merge_integration_test.go +++ b/go/buildstep/merge_integration_test.go @@ -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", "prog": "\x7fELF\x02\x01"}) + 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()) @@ -115,8 +139,8 @@ func TestMergeIntoACleanMergeAnswersTheMergeCommitAndTheChangedFiles(t *testing. 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"}}) { // issue 1577 - t.Errorf("programs = %v, want the ELF the head adds", res.Programs) + 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) diff --git a/go/buildstep/programs.go b/go/buildstep/programs.go index 8d87dce..36f47e6 100644 --- a/go/buildstep/programs.go +++ b/go/buildstep/programs.go @@ -13,6 +13,9 @@ import ( 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 @@ -40,6 +43,46 @@ func ExecutableKind(head []byte) string { // 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` (": \0\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 diff --git a/go/buildstep/programs_test.go b/go/buildstep/programs_test.go index 7cc26bc..e64dae2 100644 --- a/go/buildstep/programs_test.go +++ b/go/buildstep/programs_test.go @@ -60,8 +60,8 @@ func TestProgramsInNamesTheAddedAndModifiedRegularFilesThatArePrograms(t *testin 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 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{}}, } { From 02f50bd9d642870b7b9381c58483de101cdf572e Mon Sep 17 00:00:00 2001 From: jochen Date: Mon, 26 Oct 2026 19:52:54 +0100 Subject: [PATCH 3/4] A build seat's capacity follows its node's build CPU, and a running ask says its phases 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. 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. --- go/buildstep/askrun.go | 15 ++++ go/buildstep/capacity_cpu.go | 141 ++++++++++++++++++++++++++++++ go/buildstep/capacity_test.go | 41 ++++++++- go/buildstep/phase.go | 67 ++++++++++++++ go/buildstep/pull.go | 110 ++++++++++++++++++++--- go/buildstep/pull_test.go | 61 ++++++++++++- go/buildstep/seat_tools.go | 7 +- go/delivery/buildrun/asks.go | 17 ++++ go/delivery/buildrun/asks_test.go | 7 +- go/delivery/stepasks.go | 51 +++++++++++ go/delivery/steps.go | 5 ++ 11 files changed, 500 insertions(+), 22 deletions(-) create mode 100644 go/buildstep/capacity_cpu.go create mode 100644 go/buildstep/phase.go diff --git a/go/buildstep/askrun.go b/go/buildstep/askrun.go index d189578..0410f3a 100644 --- a/go/buildstep/askrun.go +++ b/go/buildstep/askrun.go @@ -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 { diff --git a/go/buildstep/capacity_cpu.go b/go/buildstep/capacity_cpu.go new file mode 100644 index 0000000..05123cc --- /dev/null +++ b/go/buildstep/capacity_cpu.go @@ -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.` 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 +} diff --git a/go/buildstep/capacity_test.go b/go/buildstep/capacity_test.go index a9d1a60..c52f15e 100644 --- a/go/buildstep/capacity_test.go +++ b/go/buildstep/capacity_test.go @@ -37,8 +37,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 +49,39 @@ 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)) + } } diff --git a/go/buildstep/phase.go b/go/buildstep/phase.go new file mode 100644 index 0000000..5ca842c --- /dev/null +++ b/go/buildstep/phase.go @@ -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 +} diff --git a/go/buildstep/pull.go b/go/buildstep/pull.go index 960d02e..87b95ee 100644 --- a/go/buildstep/pull.go +++ b/go/buildstep/pull.go @@ -77,6 +77,11 @@ 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) // 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 +97,11 @@ type Puller struct { // watched is the capacity the config watch handed last, used in place of Capacity once hasWatched. watched int hasWatched bool + // judged is the capacity said last (log and capacity.), 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 +226,85 @@ 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.` 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 + p.sayCapacity(c) + return c +} + +// 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 + 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 @@ -395,6 +467,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 +498,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 +515,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) } diff --git a/go/buildstep/pull_test.go b/go/buildstep/pull_test.go index 902a460..fca0690 100644 --- a/go/buildstep/pull_test.go +++ b/go/buildstep/pull_test.go @@ -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.; 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 diff --git a/go/buildstep/seat_tools.go b/go/buildstep/seat_tools.go index 07af7eb..e56d54b 100644 --- a/go/buildstep/seat_tools.go +++ b/go/buildstep/seat_tools.go @@ -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) { diff --git a/go/delivery/buildrun/asks.go b/go/delivery/buildrun/asks.go index 59d99d5..a03bbbf 100644 --- a/go/delivery/buildrun/asks.go +++ b/go/delivery/buildrun/asks.go @@ -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 { diff --git a/go/delivery/buildrun/asks_test.go b/go/delivery/buildrun/asks_test.go index 0f8540a..6905d2e 100644 --- a/go/delivery/buildrun/asks_test.go +++ b/go/delivery/buildrun/asks_test.go @@ -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"} diff --git a/go/delivery/stepasks.go b/go/delivery/stepasks.go index cba0858..33ac861 100644 --- a/go/delivery/stepasks.go +++ b/go/delivery/stepasks.go @@ -693,3 +693,54 @@ 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.` (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"` + // 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 +} diff --git a/go/delivery/steps.go b/go/delivery/steps.go index 917278a..095b1f4 100644 --- a/go/delivery/steps.go +++ b/go/delivery/steps.go @@ -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 { From d00dac140b0396d691d60c480f2171ae7ee35bc3 Mon Sep 17 00:00:00 2001 From: jochen Date: Mon, 26 Oct 2026 21:19:16 +0100 Subject: [PATCH 4/4] A build module takes asks only within its node's build hours The operator, 2026-10-26 21:18: g14 builds only from 18:00 to 06:00. A build-hours setting names a window per node in its local time (crossing midnight allowed, absent = any hour); outside it the capacity judged by is 0, running steps finish, and the puller wakes on one timer at the next edge. The change is said in the log and in capacity. with the window and when it opens. --- go/buildstep/buildhours.go | 135 ++++++++++++++++++++++++++++++++++ go/buildstep/capacity_test.go | 50 +++++++++++++ go/buildstep/pull.go | 106 +++++++++++++++++++++++++- go/delivery/stepasks.go | 6 ++ 4 files changed, 295 insertions(+), 2 deletions(-) create mode 100644 go/buildstep/buildhours.go diff --git a/go/buildstep/buildhours.go b/go/buildstep/buildhours.go new file mode 100644 index 0000000..e0bc991 --- /dev/null +++ b/go/buildstep/buildhours.go @@ -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: `=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 =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, "" + } +} diff --git a/go/buildstep/capacity_test.go b/go/buildstep/capacity_test.go index c52f15e..c44bbb5 100644 --- a/go/buildstep/capacity_test.go +++ b/go/buildstep/capacity_test.go @@ -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 @@ -84,4 +86,52 @@ func TestNodeLimitsAndTheCPUBound(t *testing.T) { 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) + } + } } diff --git a/go/buildstep/pull.go b/go/buildstep/pull.go index 87b95ee..893d631 100644 --- a/go/buildstep/pull.go +++ b/go/buildstep/pull.go @@ -82,6 +82,10 @@ type Puller struct { // (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 @@ -97,6 +101,11 @@ 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.), so a change is said once. judged delivery.Capacity hasJudged bool @@ -265,16 +274,59 @@ func (p *Puller) CapacityDetail() delivery.Capacity { 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.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 } @@ -320,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 @@ -332,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() @@ -560,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 { diff --git a/go/delivery/stepasks.go b/go/delivery/stepasks.go index 33ac861..6b8956a 100644 --- a/go/delivery/stepasks.go +++ b/go/delivery/stepasks.go @@ -724,6 +724,12 @@ type Capacity struct { // 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"`