Merge branch 'feat/r4g-asks' into feat/r4g

This commit is contained in:
2026-10-23 16:19:24 +02:00
30 changed files with 3090 additions and 109 deletions
+409
View File
@@ -0,0 +1,409 @@
package buildstep
import (
"context"
"encoding/json"
"fmt"
"path"
"slices"
"sort"
"strings"
"git.novox.be/novox/mesh-sdk/go/delivery"
)
// **One ask, one whole job** (novox/hq ADR 0525 §3). A build module runs a pulled ask with its own step tools' bodies,
// the same ones its request and reply tools run (ADR 0525 §8), in the order the job names: a build ask fetches, then
// builds or copies each artifact; a check ask fetches, digests the level, takes the kept verdicts and tests the rest,
// the packages whose tests the change touches first. The answers are the tools' own, whole, so the asker judges them as
// it judged the tools' answers.
// StepFunc is one step tool's body: its arguments in, its answer out.
type StepFunc func(ctx context.Context, args map[string]any) (any, error)
// AskSteps are the step tools a build module runs an ask with. Copy is the image module's alone; Digest the Go module's.
type AskSteps struct {
Fetch, Build, Copy, Digest, Test StepFunc
// Toolchain is the toolchain the module builds with now; a build ask naming another is refused. Nil checks nothing
// (an image or an archive names none).
Toolchain func(ctx context.Context) (delivery.ToolchainUsed, error)
// Kept looks up a kept verdict by level and digest (the checker's kept bucket, delivery.KeptKey): whether one is
// kept and how many tests it counted. Nil takes none.
Kept func(ctx context.Context, level, digest string) (tests int, found bool, err error)
// Release removes a workspace once the ask's steps on it ended (the module's release tool's body); nil leaves it to
// the module's sweep.
Release func(workspace string) error
Logf func(format string, args ...any)
}
// RunAsk runs ask's job with steps and answers a delivery.BuildAnswer or a delivery.CheckAnswer.
func RunAsk(ctx context.Context, steps AskSteps, ask delivery.StepAsk) (any, error) {
if ask.Build != nil {
return steps.build(ctx, *ask.Build)
}
if ask.Check != nil {
return steps.check(ctx, *ask.Check)
}
return nil, Refuse("the ask %s holds no job", ask.ID)
}
func raw(v any) (json.RawMessage, error) {
if r, ok := v.(json.RawMessage); ok {
return r, nil
}
return json.Marshal(v)
}
// fetched is what an ask reads of a fetch's answer.
type fetched struct {
Workspace string `json:"workspace"`
Merge string `json:"merge"`
Conflicts []string `json:"conflicts"`
Changed []string `json:"changed"`
Modules []string `json:"modules"`
}
func (s AskSteps) fetch(ctx context.Context, args map[string]any) (json.RawMessage, fetched, error) {
if s.Fetch == nil {
return nil, fetched{}, Refuse("this build module fetches nothing")
}
out, err := s.Fetch(ctx, args)
if err != nil {
return nil, fetched{}, err
}
r, err := raw(out)
if err != nil {
return nil, fetched{}, err
}
var f fetched
if err := json.Unmarshal(r, &f); err != nil || f.Workspace == "" {
return r, f, fmt.Errorf("the fetch answered no workspace: %s", r)
}
return r, f, nil
}
func (s AskSteps) build(ctx context.Context, job delivery.BuildJob) (any, error) {
answer := delivery.BuildAnswer{}
if job.Toolchain != nil && s.Toolchain != nil {
tc, err := s.Toolchain(ctx)
if err != nil {
return nil, err
}
if tc.Version != job.Toolchain.Version || tc.Recipe != job.Toolchain.Recipe {
return nil, Refuse("the ask is for %s (recipe %s) and this node builds with %s (recipe %s)",
job.Toolchain.Version, job.Toolchain.Recipe, tc.Version, tc.Recipe)
}
answer.Toolchain = &tc
}
var f fetched
defer func() { s.release(f.Workspace) }()
for _, a := range job.Artifacts {
if a.From == "" && f.Workspace == "" {
r, got, err := s.fetch(ctx, job.Fetch)
if err != nil {
return nil, err
}
if _, candidate := job.Fetch["base"]; candidate {
// A candidate is built from its head merged onto the trunk; a head that does not merge leaves the tree at the
// trunk, and building that would store the trunk's build under the candidate's fingerprint.
if len(got.Conflicts) > 0 {
return nil, fmt.Errorf("the head does not merge into the trunk: files in conflict: %s", strings.Join(got.Conflicts, ", "))
}
if got.Merge == "" {
return nil, fmt.Errorf("the fetch with a base answered no merge: %s", r)
}
}
f = got
}
var (
out any
err error
)
if a.From != "" {
if s.Copy == nil {
return nil, Refuse("this build module copies no artifact (%s)", a.Artifact)
}
out, err = s.Copy(ctx, map[string]any{"module": job.Module, "artifact": a.Artifact, "from": a.From})
} else {
if s.Build == nil {
return nil, Refuse("this build module builds no artifact (%s)", a.Artifact)
}
args := map[string]any{"workspace": f.Workspace, "module": job.Path, "artifact": a.Artifact}
if job.Publish {
args["publish"] = true
}
out, err = s.Build(ctx, args)
}
if err != nil {
return nil, fmt.Errorf("%s: %w", a.Artifact, err)
}
r, err := raw(out)
if err != nil {
return nil, err
}
var built struct {
Digest string `json:"digest"`
Reference string `json:"reference"`
}
if err := json.Unmarshal(r, &built); err != nil || built.Digest == "" {
return nil, fmt.Errorf("the build of %s answered no digest: %s", a.Artifact, r)
}
answer.Artifacts = append(answer.Artifacts, delivery.AskBuilt{Artifact: a.Artifact, Digest: built.Digest, Reference: built.Reference})
}
return answer, nil
}
func (s AskSteps) check(ctx context.Context, job delivery.CheckJob) (any, error) {
r, f, err := s.fetch(ctx, job.Fetch)
if err != nil {
return nil, err
}
defer s.release(f.Workspace)
answer := delivery.CheckAnswer{Fetch: r}
if len(f.Conflicts) > 0 {
return answer, nil // the asker reads the conflict from the fetch's answer: a verdict, not an error
}
if s.Test == nil {
return nil, Refuse("this build module tests nothing")
}
runner := ""
if job.Runner != "" && (job.RunnerWhen == "" || changesUnder(f.Changed, job.RunnerWhen)) {
runner = job.Runner
}
for _, level := range job.Levels {
la := delivery.LevelAnswer{Level: level}
passed, err := s.level(ctx, job, level, f, runner, &la)
answer.Levels = append(answer.Levels, la)
if err != nil {
return nil, err
}
if !passed {
return answer, nil // the first level that did not pass ends the check (ADR 0484 clause 13)
}
}
return answer, nil
}
// level runs one level of a check ask and says whether it passed.
func (s AskSteps) level(ctx context.Context, job delivery.CheckJob, level string, f fetched, runner string, la *delivery.LevelAnswer) (bool, error) {
withRunner := func(args map[string]any) map[string]any {
if runner != "" {
args["runner"] = runner
}
return args
}
test := func(args map[string]any) (bool, error) {
out, err := s.Test(ctx, args)
if err != nil {
return false, err
}
r, err := raw(out)
if err != nil {
return false, err
}
la.Tests = append(la.Tests, r)
var t struct {
Passed bool `json:"passed"`
}
_ = json.Unmarshal(r, &t)
return t.Passed, nil
}
if s.Digest == nil {
// A module with no digest (Node) tests each module directory of the tree, those whose tests changed first.
for _, priority := range []bool{true, false} {
for _, dir := range f.Modules {
if changesTestIn(f.Changed, dir) != priority {
continue
}
passed, err := test(withRunner(map[string]any{"workspace": f.Workspace, "module": dir, "level": level}))
if err != nil {
return false, fmt.Errorf("the %s level of %s: %w", level, dir, err)
}
if !passed {
return false, nil
}
}
}
return true, nil
}
todo, all, keepRunner := s.plan(ctx, job, level, f, runner, withRunner, la)
if !keepRunner {
runner = "" // an installed module that ignores the runner judges the level as it judges any other (ADR 0535 §2)
}
for _, priority := range []bool{true, false} {
args := withRunner(map[string]any{"workspace": f.Workspace, "level": level})
if job.Run != "" {
args["run"] = job.Run
}
if all {
if priority {
continue // nothing to put first: the level runs whole after
}
} else {
var run []string
for _, p := range todo {
if changesTestOf(f.Changed, p) == priority {
run = append(run, p)
}
}
if len(run) == 0 {
continue
}
args["packages"] = run
}
passed, err := test(args)
if err != nil {
return false, fmt.Errorf("the %s level: %w", level, err)
}
if !passed {
return false, nil
}
}
return true, nil
}
// release removes an ask's workspace, saying a failure: the module's sweep removes it later.
func (s AskSteps) release(workspace string) {
if workspace == "" || s.Release == nil {
return
}
if err := s.Release(workspace); err != nil && s.Logf != nil {
s.Logf("the workspace %s is left to the sweep: %v", workspace, err)
}
}
// plan digests the level and takes the kept verdicts: the packages left to test, or all when the level runs whole, and
// whether the runner stays. A try of some tests runs whole, digesting nothing (a partial run proves no package). Under the
// candidate's runner the digest is asked as the candidate: when the candidate answers it, every package runs and nothing
// is digested for keeping (ADR 0535 §4); when the installed module answers it (it does not take the runner), the level is
// planned as any other and the runner is dropped. The level runs whole and takes nothing when the digest names no package
// or does not cover the siblings the fetch placed (ADR 0444, L5).
func (s AskSteps) plan(ctx context.Context, job delivery.CheckJob, level string, f fetched, runner string, withRunner func(map[string]any) map[string]any, la *delivery.LevelAnswer) ([]string, bool, bool) {
if job.Run != "" {
return nil, true, true
}
out, err := s.Digest(ctx, withRunner(map[string]any{"workspace": f.Workspace, "level": level}))
if err != nil {
if s.Logf != nil {
s.Logf("no digest of the %s level, so it runs whole: %v", level, err)
}
return nil, true, true
}
r, err := raw(out)
if err != nil {
return nil, true, true
}
var d struct {
Packages map[string]string `json:"packages"`
Beside []string `json:"beside"`
JudgedBy string `json:"judged_by"`
}
if err := json.Unmarshal(r, &d); err != nil {
return nil, true, true
}
if runner != "" {
if strings.HasPrefix(d.JudgedBy, "candidate ") {
la.Judged = d.JudgedBy
return nil, true, true // the candidate decides which verdicts would be taken: all run, nothing kept
}
runner = ""
}
if len(d.Packages) == 0 || !slices.Equal(d.Beside, besideOf(job.Fetch)) {
return nil, true, runner != "" // a digest that names nothing or not the siblings keys no verdict: the level runs whole
}
la.Digest = r
pkgs := make([]string, 0, len(d.Packages))
for p := range d.Packages {
pkgs = append(pkgs, p)
}
sort.Strings(pkgs)
var todo []string
for _, p := range pkgs {
if job.Kept && s.Kept != nil {
tests, found, err := s.Kept(ctx, level, d.Packages[p])
if err != nil && s.Logf != nil {
s.Logf("the kept verdict of %s at %s cannot be read, so it runs: %v", p, level, err)
}
if err == nil && found {
la.Kept = append(la.Kept, delivery.KeptTaken{Package: p, Digest: d.Packages[p], Tests: tests})
continue
}
}
todo = append(todo, p)
}
return todo, false, false
}
// besideOf are the siblings a fetch's arguments place, as a digest echoes them: `<owner/repo> <commit>`, sorted.
func besideOf(fetch map[string]any) []string {
raw, err := json.Marshal(fetch["beside"])
if err != nil {
return nil
}
var refs []struct {
Repository string `json:"repository"`
Commit string `json:"commit"`
}
_ = json.Unmarshal(raw, &refs)
var out []string
for _, r := range refs {
out = append(out, r.Repository+" "+r.Commit)
}
sort.Strings(out)
return out
}
// changesUnder says whether a path the merge changes is under prefix.
func changesUnder(changed []string, prefix string) bool {
for _, p := range changed {
if strings.HasPrefix(p, prefix) {
return true
}
}
return false
}
// isTestFile says whether a repository-relative path is a test file: Go's _test.go, and a TypeScript module's .test.
// and .spec. files and test directories.
func isTestFile(p string) bool {
switch {
case strings.HasSuffix(p, "_test.go"):
return true
case strings.Contains(p, ".test.") || strings.Contains(p, ".spec."):
return true
}
for _, seg := range strings.Split(path.Dir(p), "/") {
if seg == "__tests__" || seg == "test" || seg == "tests" {
return true
}
}
return false
}
// changesTestOf says whether the merge changes a test file of the Go package with the import path pkg: the file's
// directory and the import path agree on their last segments.
func changesTestOf(changed []string, pkg string) bool {
for _, f := range changed {
if !strings.HasSuffix(f, "_test.go") {
continue
}
dir := path.Dir(f)
if dir == pkg || strings.HasSuffix(pkg, "/"+dir) || strings.HasSuffix(dir, "/"+pkg) {
return true
}
}
return false
}
// changesTestIn says whether the merge changes a test file inside the directory of a Node module.
func changesTestIn(changed []string, dir string) bool {
for _, f := range changed {
if !isTestFile(f) || strings.HasSuffix(f, "_test.go") {
continue
}
if dir == "." || dir == "" || strings.HasPrefix(f, strings.TrimSuffix(dir, "/")+"/") {
return true
}
}
return false
}
+205
View File
@@ -0,0 +1,205 @@
package buildstep
import (
"context"
"encoding/json"
"errors"
"fmt"
"strings"
"testing"
"time"
"git.novox.be/novox/mesh-sdk/go/delivery"
)
const runCommit = "0123456789abcdef0123456789abcdef01234567"
// recorder is a build module's step tools that record each call.
type recorder struct {
calls []string
fetch map[string]any
tests map[string]bool // package or module directory -> passed
digests map[string]string
}
func (r *recorder) steps() AskSteps {
call := func(name string, answer func(args map[string]any) any) StepFunc {
return func(ctx context.Context, args map[string]any) (any, error) {
b, _ := json.Marshal(args)
r.calls = append(r.calls, name+" "+string(b))
return answer(args), nil
}
}
return AskSteps{
Fetch: call("fetch", func(map[string]any) any { return r.fetch }),
Build: call("build", func(args map[string]any) any {
return map[string]any{"digest": "sha256:" + args["artifact"].(string), "reference": "artifact-store://" + args["artifact"].(string)}
}),
Copy: call("copy", func(args map[string]any) any { return map[string]any{"digest": "sha256:copied"} }),
Digest: call("digest", func(map[string]any) any { return map[string]any{"level": "unit", "packages": r.digests} }),
Test: call("test", func(args map[string]any) any {
passed := true
if pkgs, ok := args["packages"].([]string); ok {
for _, p := range pkgs {
passed = passed && r.tests[p]
}
}
if dir, ok := args["module"].(string); ok {
passed = r.tests[dir]
}
return map[string]any{"passed": passed, "tests": 1}
}),
Kept: func(ctx context.Context, level, digest string) (int, bool, error) { return 7, digest == "d-kept", nil },
}
}
func ask(job any) delivery.StepAsk {
a := delivery.StepAsk{ID: NewID(), Kind: delivery.AskCheck, Language: delivery.AskGo, Asker: "t", At: time.Now()}
switch j := job.(type) {
case delivery.BuildJob:
a.Build = &j
case delivery.CheckJob:
a.Check = &j
}
return a
}
// 1887/A3: a build ask fetches once, builds each artifact from the workspace with publish, copies a pinned one without a
// fetch, and answers each artifact's digest and reference.
func TestRunAskBuilds(t *testing.T) {
r := &recorder{fetch: map[string]any{"workspace": "/w/1"}}
steps := r.steps()
var released []string
steps.Release = func(ws string) error { released = append(released, ws); return nil }
out, err := RunAsk(context.Background(), steps, ask(delivery.BuildJob{Module: "m", Path: "modules/m", Publish: true,
Fetch: map[string]any{"repository": "novox/x", "commit": runCommit},
Artifacts: []delivery.AskArtifact{{Artifact: "base", From: "registry/b@sha256:00"}, {Artifact: "bundle"}, {Artifact: "lib"}}}))
if err != nil {
t.Fatal(err)
}
got := strings.Join(r.calls, "\n")
want := `copy {"artifact":"base","from":"registry/b@sha256:00","module":"m"}
fetch {"commit":"` + runCommit + `","repository":"novox/x"}
build {"artifact":"bundle","module":"modules/m","publish":true,"workspace":"/w/1"}
build {"artifact":"lib","module":"modules/m","publish":true,"workspace":"/w/1"}`
if got != want {
t.Fatalf("calls:\n%s\nwant:\n%s", got, want)
}
if len(released) != 1 || released[0] != "/w/1" {
t.Errorf("released %v, want the one workspace", released)
}
a := out.(delivery.BuildAnswer)
if len(a.Artifacts) != 3 || a.Artifacts[1].Reference != "artifact-store://bundle" {
t.Fatalf("answer %+v", a)
}
}
// A build ask for another toolchain is refused, so its step ends refused and the asker reads why.
func TestRunAskRefusesAnotherToolchain(t *testing.T) {
r := &recorder{fetch: map[string]any{"workspace": "/w"}}
s := r.steps()
s.Toolchain = func(context.Context) (delivery.ToolchainUsed, error) {
return delivery.ToolchainUsed{Version: "go1.27", Recipe: "3"}, nil
}
_, err := RunAsk(context.Background(), s, ask(delivery.BuildJob{Module: "m", Path: ".", Fetch: map[string]any{"repository": "novox/x", "commit": runCommit},
Artifacts: []delivery.AskArtifact{{Artifact: "bundle"}}, Toolchain: &delivery.ToolchainUsed{Version: "go1.26", Recipe: "3"}}))
var ref *Refusal
if !errors.As(err, &ref) || len(r.calls) != 0 {
t.Fatalf("err %v, calls %v", err, r.calls)
}
}
// 1887/A3, A5a, issue 1887 design limit 1: a check ask runs the levels of one check after one fetch: it digests the unit
// level, takes the kept verdicts, tests the packages whose tests changed first, and stops at the first failure, running
// no later level.
func TestRunAskChecksTheLevelsOfOneCheck(t *testing.T) {
r := &recorder{
fetch: map[string]any{"workspace": "/w", "changed": []string{"pkg/b/b_test.go"}},
digests: map[string]string{"x/pkg/a": "d-a", "x/pkg/b": "d-b", "x/pkg/k": "d-kept"},
tests: map[string]bool{"x/pkg/a": true, "x/pkg/b": false},
}
out, err := RunAsk(context.Background(), r.steps(), ask(delivery.CheckJob{Levels: []string{delivery.LevelUnit, delivery.LevelIntegration}, Kept: true,
Runner: "workspace", RunnerWhen: "modules/mesh-build-go/", Fetch: map[string]any{"repository": "novox/x", "commit": runCommit, "base": runCommit}}))
if err != nil {
t.Fatal(err)
}
got := strings.Join(r.calls, "\n")
want := `fetch {"base":"` + runCommit + `","commit":"` + runCommit + `","repository":"novox/x"}
digest {"level":"unit","workspace":"/w"}
test {"level":"unit","packages":["x/pkg/b"],"workspace":"/w"}`
if got != want {
t.Fatalf("calls:\n%s\nwant:\n%s", got, want)
}
a := out.(delivery.CheckAnswer)
if len(a.Levels) != 1 {
t.Fatalf("levels %+v, want the unit level alone", a.Levels)
}
u := a.Levels[0]
if u.Level != delivery.LevelUnit || len(u.Kept) != 1 || u.Kept[0].Package != "x/pkg/k" || u.Kept[0].Tests != 7 || len(u.Tests) != 1 || len(u.Digest) == 0 {
t.Fatalf("the unit level %+v", u)
}
}
// ADR 0535 §2, §4: a merge that changes the build module's own code is judged by the candidate's runner: its digest says
// the candidate ran, and every package runs with nothing kept; an installed module that ignores the runner is planned as
// any other, the runner dropped. A digest that does not cover the fetch's siblings runs the level whole too.
func TestRunAskRunsWholeUnderTheRunnerOrWithoutTheSiblings(t *testing.T) {
for _, tc := range []struct{ judged, want string }{
{"candidate 999", `digest {"level":"unit","runner":"workspace","workspace":"/w"}
test {"level":"unit","runner":"workspace","workspace":"/w"}`},
{"installed", `digest {"level":"unit","runner":"workspace","workspace":"/w"}`},
} {
r := &recorder{fetch: map[string]any{"workspace": "/w", "changed": []string{"modules/mesh-build-go/x.go"}},
digests: map[string]string{"x/pkg/a": "d-kept"}, tests: map[string]bool{}}
steps := r.steps()
digest := steps.Digest
steps.Digest = func(ctx context.Context, args map[string]any) (any, error) {
_, _ = digest(ctx, args)
return map[string]any{"level": "unit", "packages": r.digests, "judged_by": tc.judged}, nil
}
out, err := RunAsk(context.Background(), steps, ask(delivery.CheckJob{Levels: []string{delivery.LevelUnit}, Kept: true,
Runner: "workspace", RunnerWhen: "modules/mesh-build-go/", Fetch: map[string]any{"repository": "novox/x", "commit": runCommit}}))
if err != nil {
t.Fatal(err)
}
if got := strings.Join(r.calls[1:], "\n"); got != tc.want {
t.Errorf("judged by %s: calls\n%s\nwant\n%s", tc.judged, got, tc.want)
}
l := out.(delivery.CheckAnswer).Levels[0]
if tc.judged == "installed" && (len(l.Kept) != 1 || l.Judged != "") {
t.Errorf("an installed module's level %+v, want the kept verdict taken", l)
}
if tc.judged != "installed" && (len(l.Kept) != 0 || len(l.Digest) != 0 || l.Judged != tc.judged) {
t.Errorf("the candidate's level %+v, want nothing kept and the program named", l)
}
}
r := &recorder{fetch: map[string]any{"workspace": "/w"}, digests: map[string]string{"x/pkg/a": "d-kept"}, tests: map[string]bool{}}
out, err := RunAsk(context.Background(), r.steps(), ask(delivery.CheckJob{Levels: []string{delivery.LevelUnit}, Kept: true,
Fetch: map[string]any{"repository": "novox/x", "commit": runCommit, "beside": []any{map[string]any{"repository": "novox/mesh-sdk", "commit": runCommit}}}}))
if err != nil {
t.Fatal(err)
}
if l := out.(delivery.CheckAnswer).Levels[0]; len(l.Kept) != 0 || len(l.Digest) != 0 {
t.Fatalf("a digest without the siblings kept %+v", l)
}
}
// A Node check tests each module directory, the one whose tests changed first; a conflict ends the ask with the fetch
// alone, a verdict for the asker.
func TestRunAskChecksNodeModulesAndConflicts(t *testing.T) {
r := &recorder{fetch: map[string]any{"workspace": "/w", "modules": []string{"a", "b"}, "changed": []string{"b/test/x.test.ts"}},
tests: map[string]bool{"a": true, "b": true}}
s := r.steps()
s.Digest = nil
if _, err := RunAsk(context.Background(), s, ask(delivery.CheckJob{Levels: []string{delivery.LevelUnit}, Fetch: map[string]any{"repository": "novox/x", "commit": runCommit}})); err != nil {
t.Fatal(err)
}
if fmt.Sprint(r.calls[1:]) != `[test {"level":"unit","module":"b","workspace":"/w"} test {"level":"unit","module":"a","workspace":"/w"}]` {
t.Fatalf("calls %v", r.calls)
}
r = &recorder{fetch: map[string]any{"workspace": "/w", "conflicts": []string{"go.mod"}}}
out, err := RunAsk(context.Background(), r.steps(), ask(delivery.CheckJob{Levels: []string{delivery.LevelUnit}, Fetch: map[string]any{"repository": "novox/x", "commit": runCommit}}))
if err != nil || len(r.calls) != 1 || len(out.(delivery.CheckAnswer).Levels) != 0 {
t.Fatalf("a conflict: %v %v", err, r.calls)
}
}
+52
View File
@@ -0,0 +1,52 @@
package buildstep
import (
"encoding/json"
"fmt"
"os"
"git.novox.be/novox/mesh-sdk/go/delivery/buildrun"
)
// CapacitySetting is the build modules' setting of each node's build capacity (novox/hq ADR 0525 §5): `<node>=<asks at
// once>` pairs separated by commas, as the builder's was (ADR 0485 §1). Each build module declares it with
// DefaultCapacities as its default, and the controller lays it into the module's settings file.
const CapacitySetting = "capacity"
// DefaultCapacities is the setting's default, the builder's before R4g: the control-node (novox) and the laptop (g14)
// build nothing.
const DefaultCapacities = "shanks=3,ace=2,g14=0,novox=0"
// NodeCapacity is node's build capacity, read from the module's settings file at path at every call, so a change takes
// effect at the next pull; with no path it reads fallback. A node the setting does not name runs one ask at a time (as
// the builder's did). A file or a setting that cannot be read gives 0 and says why: pulling on a setting nobody can read
// could build on a node that must not.
func NodeCapacity(path, fallback, node string, logf func(string, ...any)) func() int {
say := func(format string, args ...any) {
if logf != nil {
logf(format, args...)
}
}
return func() int {
setting := fallback
if path != "" {
raw, err := os.ReadFile(path)
if err != nil {
say("the build capacity cannot be read, so %s pulls nothing: %v", node, err)
return 0
}
var s map[string]any
if err := json.Unmarshal(raw, &s); err != nil {
say("the settings file %s is not JSON, so %s pulls nothing: %v", path, node, err)
return 0
}
setting, _ = s[CapacitySetting].(string)
}
caps, err := buildrun.ParseCapacity(setting)
if err != nil {
say("%v, so %s pulls nothing", fmt.Errorf("the build capacity: %w", err), node)
return 0
}
return (&buildrun.Places{Capacity: caps}).CapacityOf(node)
}
}
+38
View File
@@ -0,0 +1,38 @@
package buildstep
import (
"os"
"path/filepath"
"testing"
)
// 1887/A7: the node's build capacity is the build module's setting, read at every pull: the control-node at 0 pulls
// nothing, a change takes effect at the next read, and a setting that cannot be read pulls nothing.
func TestNodeCapacity(t *testing.T) {
if got := NodeCapacity("", DefaultCapacities, "novox", nil)(); got != 0 {
t.Errorf("novox = %d by default, want 0", got)
}
if got := NodeCapacity("", DefaultCapacities, "shanks", nil)(); got != 3 {
t.Errorf("shanks = %d by default, want 3", got)
}
file := filepath.Join(t.TempDir(), "settings.json")
os.WriteFile(file, []byte(`{"capacity":"shanks=1,ace=2"}`), 0o600)
capacity := NodeCapacity(file, DefaultCapacities, "shanks", nil)
if got := capacity(); got != 1 {
t.Errorf("shanks = %d from the file, want 1", got)
}
os.WriteFile(file, []byte(`{"capacity":"shanks=4"}`), 0o600)
if got := capacity(); got != 4 {
t.Errorf("shanks = %d after the change, want 4", got)
}
if got := NodeCapacity(file, "", "pi", nil)(); got != 1 {
t.Errorf("a node not named = %d, want 1", got)
}
os.WriteFile(file, []byte(`{"capacity":"shanks=lots"}`), 0o600)
if got := capacity(); got != 0 {
t.Errorf("an unreadable setting = %d, want 0", got)
}
if got := NodeCapacity(filepath.Join(t.TempDir(), "none"), DefaultCapacities, "shanks", nil)(); got != 0 {
t.Errorf("a missing file = %d, want 0", got)
}
}
+335
View File
@@ -0,0 +1,335 @@
package buildstep
import (
"context"
"encoding/json"
"errors"
"fmt"
"sort"
"sync"
"time"
"git.novox.be/novox/mesh-sdk/go/delivery"
)
// **The pull loop the build modules share** (novox/hq ADR 0525 §4-§6, ADR 0536 §4; issue 1887). A build module on a node
// whose build capacity is above zero pulls the asks of its language's seat: from the delivery worker first, then check,
// then recheck, holding at most its capacity at once. Each ask runs as a step under the ask's own id, so its end is the
// key <node>.<ask id> in the module's steps state, which the asker watches; the ask is acknowledged when that end is
// written, kept alive while it runs by an in-progress signal, and redelivered by the stream to another node if this one
// stops. No part keeps a timer for an ask: the pull's own wait and the stream's acknowledgement wait are the bus's.
// Pulled is one ask a source handed over, with what decides its fate on the bus.
type Pulled struct {
// Kind is the worker it came from, filled by the puller.
Kind string
Data []byte
// Delivered is how many times the bus has handed it over, this time included.
Delivered uint64
// Ack ends it: it is never delivered again. InProgress tells the bus it is still being worked, restarting its
// acknowledgement wait. Nak hands it back to be delivered again at once (to this node or another). Term drops it.
Ack, InProgress, Nak, Term func() error
}
// AskSource is where a puller takes asks from: the seat's workers on the bus (JetStreamAsks) or a test double
// (FakeAsks).
type AskSource interface {
// Next takes one ask of kind: with wait 0 only one already waiting, else waiting for one at most wait. It answers
// nil and no error when none came.
Next(ctx context.Context, kind string, wait time.Duration) (*Pulled, error)
}
// Defaults of a Puller.
var (
// DefaultPullWait is how long one pull of the delivery worker waits for an ask while every worker is empty: the
// pull's own wait (as process.Consume's), after which the check and recheck workers are looked at again.
DefaultPullWait = 5 * time.Second
// DefaultHeartbeat is how often a running ask tells the bus it is still being worked; under the seat worker's
// acknowledgement wait of 60 s (mesh-controller internal/broker HolderConsumerFor).
DefaultHeartbeat = 20 * time.Second
)
// Puller pulls a build seat's asks and runs them.
type Puller struct {
Source AskSource
Language string // the seat's language: delivery.AskGo, AskNode or AskImage
Node string // the node this holder runs on, the first part of its step keys
Put Put // writes the module's steps state
// Capacity is how many asks the node runs at once; read before every pull, so a change takes effect at the next
// one. 0 pulls nothing.
Capacity func() int
// Run runs one ask's job; a *Refusal ends its step refused. Its result is the step's result.
Run func(ctx context.Context, ask delivery.StepAsk) (any, error)
// PullWait and Heartbeat default to DefaultPullWait and DefaultHeartbeat.
PullWait time.Duration
Heartbeat time.Duration
Logf func(format string, args ...any)
// Load reads one key of the module's steps state (nil and no error when it is absent). When set, Pull reads the
// node's pause key (delivery.PauseKey) before it pulls, and Pause and Resume write it with Put, so a pause outlives
// a restart.
Load func(key string) ([]byte, error)
mu sync.Mutex
inHand int
running map[string]RunningAsk
paused bool
wake chan struct{}
once sync.Once
}
func (p *Puller) init() {
p.once.Do(func() { p.wake = make(chan struct{}, 1) })
}
func (p *Puller) logf(format string, args ...any) {
if p.Logf != nil {
p.Logf(format, args...)
}
}
// Wake makes a waiting puller look again: after a capacity change, a resume, or a step's end.
func (p *Puller) Wake() {
p.init()
select {
case p.wake <- struct{}{}:
default:
}
}
// Pause stops the puller taking asks; the asks it holds run to their end. by names who paused it. With Load set the
// pause is written on the node's pause key first, and an error there leaves the puller as it was.
func (p *Puller) Pause(by string) error { return p.setPaused(true, by) }
// Resume lets the puller take asks again, written as Pause's is.
func (p *Puller) Resume(by string) error { return p.setPaused(false, by) }
func (p *Puller) setPaused(paused bool, by string) error {
if p.Load != nil {
key, err := delivery.PauseKey(p.Node)
if err != nil {
return err
}
raw, err := delivery.EncodePause(delivery.Pause{Paused: paused, By: by, At: now()})
if err != nil {
return err
}
if _, err := p.Put(key, json.RawMessage(raw)); err != nil {
return fmt.Errorf("buildstep: writing the pause of %s: %w", p.Node, err)
}
}
p.mu.Lock()
p.paused = paused
p.mu.Unlock()
p.Wake()
return nil
}
// restore reads the node's pause key, when Load is set.
func (p *Puller) restore() error {
if p.Load == nil {
return nil
}
key, err := delivery.PauseKey(p.Node)
if err != nil {
return err
}
raw, err := p.Load(key)
if err != nil {
return fmt.Errorf("buildstep: reading the pause of %s: %w", p.Node, err)
}
if raw == nil {
return nil
}
v, err := delivery.DecodePause(raw)
if err != nil {
return fmt.Errorf("buildstep: the pause of %s: %w", p.Node, err)
}
p.mu.Lock()
p.paused = v.Paused
p.mu.Unlock()
return nil
}
// Paused says whether the puller takes no asks.
func (p *Puller) Paused() bool {
p.mu.Lock()
defer p.mu.Unlock()
return p.paused
}
// InHand is how many asks run now.
func (p *Puller) InHand() int {
p.mu.Lock()
defer p.mu.Unlock()
return p.inHand
}
// RunningAsk is an ask a puller runs now, as its seat's current verb says it.
type RunningAsk struct {
ID string `json:"id"`
Kind string `json:"kind"`
Module string `json:"module,omitempty"`
Started time.Time `json:"started"`
}
// Running are the asks the puller runs now, oldest first.
func (p *Puller) Running() []RunningAsk {
p.mu.Lock()
defer p.mu.Unlock()
out := make([]RunningAsk, 0, len(p.running))
for _, r := range p.running {
out = append(out, r)
}
sort.Slice(out, func(i, j int) bool { return out[i].Started.Before(out[j].Started) })
return out
}
// room says whether the puller may take one more ask now.
func (p *Puller) room() bool {
capacity := 0
if p.Capacity != nil {
capacity = p.Capacity()
}
p.mu.Lock()
defer p.mu.Unlock()
return !p.paused && p.inHand < capacity
}
// Pull takes and runs asks until ctx ends. An ask still running then is neither ended nor acknowledged: the stream
// delivers it again, to another node or to this one restarted.
func (p *Puller) Pull(ctx context.Context) error {
p.init()
if p.Source == nil || p.Put == nil || p.Run == nil {
return errors.New("buildstep: a puller needs its source, its steps' put and its run")
}
if err := p.restore(); err != nil {
return err // pulling on a pause that cannot be read could build on a node paused on purpose
}
wait := p.PullWait
if wait <= 0 {
wait = DefaultPullWait
}
for {
if err := ctx.Err(); err != nil {
return err
}
if !p.room() {
select {
case <-ctx.Done():
return ctx.Err()
case <-p.wake:
}
continue
}
msg, err := p.next(ctx, wait)
if err != nil {
if ctx.Err() != nil {
return ctx.Err()
}
p.logf("the %s asks cannot be pulled: %v", p.Language, err)
select { // the pull's own wait before the next try, as an empty pull's
case <-ctx.Done():
return ctx.Err()
case <-p.wake:
case <-time.After(wait):
}
continue
}
if msg == nil {
continue
}
p.take(ctx, msg)
}
}
// next is one ask from the workers in their order: one already waiting on delivery, check or recheck, and when none
// waits, a pull of the delivery worker that waits for one.
func (p *Puller) next(ctx context.Context, wait time.Duration) (*Pulled, error) {
for _, kind := range delivery.AskKinds {
msg, err := p.Source.Next(ctx, kind, 0)
if err != nil || msg != nil {
return withKind(msg, kind), err
}
}
msg, err := p.Source.Next(ctx, delivery.AskDelivery, wait)
return withKind(msg, delivery.AskDelivery), err
}
func withKind(msg *Pulled, kind string) *Pulled {
if msg != nil {
msg.Kind = kind
}
return msg
}
// take starts one ask: it decodes it, refuses one that is not of this seat, and runs it as a step under its own id.
func (p *Puller) take(ctx context.Context, msg *Pulled) {
kind := msg.Kind
ask, err := delivery.DecodeStepAsk(msg.Data)
if err == nil && (ask.Language != p.Language || (kind != "" && ask.Kind != kind)) {
err = fmt.Errorf("the ask %s is a %s ask of %s, pulled from the %s %s worker", ask.ID, ask.Kind, ask.Language, p.Language, kind)
}
if err != nil {
// An ask that is not one runs nowhere: delivering it again would only fail again.
p.logf("an ask dropped: %v", err)
_ = msg.Term()
return
}
step, key, err := begin(p.Put, p.Node, ask.Kind+" ask", ask.ID)
if err != nil {
p.logf("the ask %s cannot start on %s: %v", ask.ID, p.Node, err)
_ = msg.Nak()
return
}
module := ""
if ask.Build != nil {
module = ask.Build.Module
}
p.mu.Lock()
p.inHand++
if p.running == nil {
p.running = map[string]RunningAsk{}
}
p.running[ask.ID] = RunningAsk{ID: ask.ID, Kind: ask.Kind, Module: module, Started: step.Started}
p.mu.Unlock()
go p.run(ctx, msg, ask, key, step)
}
func (p *Puller) run(ctx context.Context, msg *Pulled, ask delivery.StepAsk, key string, step delivery.Step) {
defer func() {
p.mu.Lock()
p.inHand--
delete(p.running, ask.ID)
p.mu.Unlock()
p.Wake()
}()
beat := p.Heartbeat
if beat <= 0 {
beat = DefaultHeartbeat
}
done := make(chan struct{})
go func() {
t := time.NewTicker(beat)
defer t.Stop()
for {
select {
case <-done:
return
case <-t.C:
if msg.InProgress != nil {
_ = msg.InProgress()
}
}
}
}()
result, err := protect(ctx, func(ctx context.Context) (any, error) { return p.Run(ctx, ask) })
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.
return
}
finish(p.Put, key, step, result, err)
if aerr := msg.Ack(); aerr != nil {
p.logf("the ask %s ended on %s and its acknowledgement failed: %v", ask.ID, p.Node, aerr)
}
}
+102
View File
@@ -0,0 +1,102 @@
package buildstep
import (
"context"
"fmt"
"sync"
"time"
"github.com/nats-io/nats.go/jetstream"
"git.novox.be/novox/mesh-sdk/go/delivery"
)
// JetStreamAsks is a build seat's workers on the bus: the shared pull consumer of each kind on the seat's stream
// (delivery.AskStream, AskWorker), which the broker makes when the seat is declared; a holder only binds them.
type JetStreamAsks struct {
JS jetstream.JetStream
Language string
mu sync.Mutex
cons map[string]jetstream.Consumer
}
func (s *JetStreamAsks) consumer(ctx context.Context, kind string) (jetstream.Consumer, error) {
s.mu.Lock()
defer s.mu.Unlock()
if c, ok := s.cons[kind]; ok {
return c, nil
}
stream, err := delivery.AskStream(s.Language)
if err != nil {
return nil, err
}
worker, err := delivery.AskWorker(s.Language, kind)
if err != nil {
return nil, err
}
c, err := s.JS.Consumer(ctx, stream, worker)
if err != nil {
return nil, fmt.Errorf("binding the %s worker %s on %s: %w", kind, worker, stream, err)
}
if s.cons == nil {
s.cons = map[string]jetstream.Consumer{}
}
s.cons[kind] = c
return c, nil
}
// Next is AskSource's: a fetch of one with no wait when wait is 0, else one that waits at most wait.
func (s *JetStreamAsks) Next(ctx context.Context, kind string, wait time.Duration) (*Pulled, error) {
c, err := s.consumer(ctx, kind)
if err != nil {
return nil, err
}
var batch jetstream.MessageBatch
if wait <= 0 {
batch, err = c.FetchNoWait(1)
} else {
batch, err = c.Fetch(1, jetstream.FetchMaxWait(wait))
}
if err != nil {
return nil, err
}
select {
case msg, ok := <-batch.Messages():
if !ok {
return nil, batch.Error()
}
var delivered uint64 = 1
if meta, err := msg.Metadata(); err == nil && meta != nil {
delivered = meta.NumDelivered
}
return &Pulled{Data: append([]byte(nil), msg.Data()...), Delivered: delivered,
Ack: msg.Ack, InProgress: msg.InProgress, Nak: msg.Nak, Term: msg.Term}, nil
case <-ctx.Done():
return nil, ctx.Err()
}
}
// DeclareAskWorkers makes language's seat stream and its three workers on a bus with no broker: a test's (the build
// modules' integration tests start one). On the mesh the broker declares them from the seat, as it declares every
// seat's stream (mesh-controller internal/broker SeatStreams); a holder never does.
func DeclareAskWorkers(ctx context.Context, js jetstream.JetStream, language string, ackWait time.Duration) error {
stream, err := delivery.AskStream(language)
if err != nil {
return err
}
seat, _ := delivery.BuildSeat(language)
if _, err := js.CreateOrUpdateStream(ctx, jetstream.StreamConfig{Name: stream, Subjects: []string{"mesh.seat." + seat + ".accept.>"},
Retention: jetstream.WorkQueuePolicy}); err != nil {
return fmt.Errorf("the stream %s: %w", stream, err)
}
for _, kind := range delivery.AskKinds {
worker, _ := delivery.AskWorker(language, kind)
subject, _ := delivery.AskSubject(language, kind)
if _, err := js.CreateOrUpdateConsumer(ctx, stream, jetstream.ConsumerConfig{Durable: worker, FilterSubject: subject,
AckPolicy: jetstream.AckExplicitPolicy, AckWait: ackWait, MaxDeliver: 5}); err != nil {
return fmt.Errorf("the worker %s: %w", worker, err)
}
}
return nil
}
+107
View File
@@ -0,0 +1,107 @@
package buildstep
import (
"context"
"sync"
"time"
"git.novox.be/novox/mesh-sdk/go/delivery"
)
// FakeAsks is a test double of a seat's workers for the build modules' unit tests: Publish puts an ask on a kind's
// worker, Next hands them out in their order, and the fates the puller gives them are counted. Safe for concurrent use.
type FakeAsks struct {
mu sync.Mutex
waiting map[string][][]byte
added chan struct{}
// Acked, Termed and Naked are the asks' ids or data by fate, InProgress how many heartbeats were given.
Acked, Termed, Naked []string
InProgress int
}
// NewFakeAsks is an empty double.
func NewFakeAsks() *FakeAsks {
return &FakeAsks{waiting: map[string][][]byte{}, added: make(chan struct{}, 1)}
}
// Publish puts an encoded ask on kind's worker.
func (f *FakeAsks) Publish(kind string, data []byte) {
f.mu.Lock()
f.waiting[kind] = append(f.waiting[kind], data)
f.mu.Unlock()
select {
case f.added <- struct{}{}:
default:
}
}
// PublishAsk encodes a and puts it on its kind's worker.
func (f *FakeAsks) PublishAsk(a delivery.StepAsk) error {
raw, err := delivery.EncodeStepAsk(a)
if err != nil {
return err
}
f.Publish(a.Kind, raw)
return nil
}
// Waiting is how many asks wait on kind's worker.
func (f *FakeAsks) Waiting(kind string) int {
f.mu.Lock()
defer f.mu.Unlock()
return len(f.waiting[kind])
}
// Fates is a copy of the acked, termed and naked lists.
func (f *FakeAsks) Fates() (acked, termed, naked []string) {
f.mu.Lock()
defer f.mu.Unlock()
return append([]string(nil), f.Acked...), append([]string(nil), f.Termed...), append([]string(nil), f.Naked...)
}
// Next is AskSource's.
func (f *FakeAsks) Next(ctx context.Context, kind string, wait time.Duration) (*Pulled, error) {
var deadline <-chan time.Time
if wait > 0 {
t := time.NewTimer(wait)
defer t.Stop()
deadline = t.C
}
for {
f.mu.Lock()
if q := f.waiting[kind]; len(q) > 0 {
data := q[0]
f.waiting[kind] = q[1:]
f.mu.Unlock()
return f.pulled(data), nil
}
f.mu.Unlock()
if wait <= 0 {
return nil, nil
}
select {
case <-ctx.Done():
return nil, ctx.Err()
case <-deadline:
return nil, nil
case <-f.added:
}
}
}
func (f *FakeAsks) pulled(data []byte) *Pulled {
id := string(data)
if a, err := delivery.DecodeStepAsk(data); err == nil {
id = a.ID
}
fate := func(list *[]string) func() error {
return func() error {
f.mu.Lock()
*list = append(*list, id)
f.mu.Unlock()
return nil
}
}
return &Pulled{Data: data, Delivered: 1, Ack: fate(&f.Acked), Nak: fate(&f.Naked), Term: fate(&f.Termed),
InProgress: func() error { f.mu.Lock(); f.InProgress++; f.mu.Unlock(); return nil }}
}
+144
View File
@@ -0,0 +1,144 @@
//go:build integration
package buildstep
import (
"context"
"strings"
"sync"
"testing"
"time"
"github.com/nats-io/nats.go"
"github.com/nats-io/nats.go/jetstream"
"git.novox.be/novox/mesh-sdk/go/delivery"
"git.novox.be/novox/mesh-sdk/go/testbus"
)
func askBus(t *testing.T, ackWait time.Duration) jetstream.JetStream {
t.Helper()
nc, err := nats.Connect(testbus.URL(t))
if err != nil {
t.Fatal(err)
}
t.Cleanup(nc.Close)
js, err := jetstream.New(nc)
if err != nil {
t.Fatal(err)
}
if err := DeclareAskWorkers(context.Background(), js, delivery.AskGo, ackWait); err != nil {
t.Fatal(err)
}
return js
}
func publishAsk(t *testing.T, js jetstream.JetStream, a delivery.StepAsk) {
t.Helper()
raw, err := delivery.EncodeStepAsk(a)
if err != nil {
t.Fatal(err)
}
subject, _ := delivery.AskSubject(a.Language, a.Kind)
if _, err := js.Publish(context.Background(), subject, raw); err != nil {
t.Fatal(err)
}
}
// 1887/A2a, 1887/AC-order, 1887/AC-pull on the real server: a check ask published before a delivery ask runs after it;
// with capacity 2 and three asks two run, the third waits in the stream and starts when one ends.
func TestPullOrderAndCapacityOnTheBus(t *testing.T) {
js := askBus(t, 30*time.Second)
check, first, second := testAsk(delivery.AskCheck), testAsk(delivery.AskDelivery), testAsk(delivery.AskDelivery)
publishAsk(t, js, check)
publishAsk(t, js, first)
publishAsk(t, js, second)
release := make(chan struct{})
var mu sync.Mutex
var order []string
st := newSteps()
p := &Puller{Source: &JetStreamAsks{JS: js, Language: delivery.AskGo}, Language: delivery.AskGo, Node: "shanks", Put: st.put,
Capacity: func() int { return 2 }, PullWait: 200 * time.Millisecond,
Run: func(ctx context.Context, a delivery.StepAsk) (any, error) {
mu.Lock()
order = append(order, a.ID)
mu.Unlock()
<-release
return nil, nil
}}
ctx, cancel := context.WithCancel(context.Background())
defer cancel()
go p.Pull(ctx)
waitFor(t, "two in hand", func() bool { return p.InHand() == 2 })
time.Sleep(500 * time.Millisecond)
mu.Lock()
if len(order) != 2 || order[0] != first.ID || order[1] != second.ID {
t.Fatalf("ran %v, want the two delivery asks %s %s first", order, first.ID, second.ID)
}
mu.Unlock()
release <- struct{}{}
waitFor(t, "the check taken", func() bool { mu.Lock(); defer mu.Unlock(); return len(order) == 3 })
close(release)
waitFor(t, "every step ended", func() bool {
for _, a := range []delivery.StepAsk{check, first, second} {
key, _ := delivery.AskStepKey("shanks", a.ID)
if s, ok := st.get(key); !ok || s.State != delivery.StepPassed {
return false
}
}
return true
})
info, err := js.Stream(ctx, "SEAT_NODE_BUILD_GO")
if err != nil {
t.Fatal(err)
}
waitFor(t, "the stream empty (every ask acknowledged)", func() bool {
si, err := info.Info(ctx)
return err == nil && si.State.Msgs == 0
})
}
// 1887/A2c, ADR 0525 §6: a killed puller's ask is redelivered by the stream's acknowledgement wait to the other node,
// whose step key names the reference.
func TestPullRedeliversAKilledPullersAsk(t *testing.T) {
js := askBus(t, 2*time.Second)
a := testAsk(delivery.AskDelivery)
publishAsk(t, js, a)
st := newSteps()
taken := make(chan struct{})
dying := &Puller{Source: &JetStreamAsks{JS: js, Language: delivery.AskGo}, Language: delivery.AskGo, Node: "ace", Put: st.put,
Capacity: func() int { return 1 }, PullWait: 200 * time.Millisecond, Heartbeat: time.Hour,
Run: func(ctx context.Context, a delivery.StepAsk) (any, error) {
close(taken)
<-ctx.Done()
return nil, ctx.Err()
}}
kill, killed := context.WithCancel(context.Background())
go dying.Pull(kill)
<-taken
killed()
other := &Puller{Source: &JetStreamAsks{JS: js, Language: delivery.AskGo}, Language: delivery.AskGo, Node: "shanks", Put: st.put,
Capacity: func() int { return 1 }, PullWait: 200 * time.Millisecond,
Run: func(ctx context.Context, a delivery.StepAsk) (any, error) {
return delivery.BuildAnswer{Artifacts: []delivery.AskBuilt{{Artifact: "bundle", Digest: "sha256:00", Reference: "artifact-store://x"}}}, nil
}}
ctx, cancel := context.WithCancel(context.Background())
defer cancel()
go other.Pull(ctx)
key, _ := delivery.AskStepKey("shanks", a.ID)
deadline := time.Now().Add(10 * time.Second)
for {
if s, ok := st.get(key); ok && s.State == delivery.StepPassed {
if want := `artifact-store://x`; !contains(string(s.Result), want) {
t.Fatalf("the step's result %s names no %s", s.Result, want)
}
return
}
if time.Now().After(deadline) {
t.Fatalf("the killed puller's ask was not taken by shanks within 10 s")
}
time.Sleep(20 * time.Millisecond)
}
}
func contains(s, sub string) bool { return strings.Contains(s, sub) }
+44
View File
@@ -0,0 +1,44 @@
package buildstep
import (
"context"
"time"
stdio "git.novox.be/novox/mesh-sdk/go"
"git.novox.be/novox/mesh-sdk/go/delivery"
)
// SeatAsks is a build seat's workers as a tools bundle reaches them: through its runtime's seat pull and settle
// (stdio.SeatPull, SeatSettle; issue 1887 A11), one message per pull, only from the workers the module's membership
// lists.
type SeatAsks struct {
Language string
// Pull and Settle are stdio.SeatPull and stdio.SeatSettle when nil; a test gives its own.
Pull func(worker string, wait time.Duration) (*stdio.Pulled, error)
Settle func(id, fate string) error
}
// Next is AskSource's.
func (s SeatAsks) Next(ctx context.Context, kind string, wait time.Duration) (*Pulled, error) {
pull, settle := s.Pull, s.Settle
if pull == nil {
pull = stdio.SeatPull
}
if settle == nil {
settle = stdio.SeatSettle
}
worker, err := delivery.AskWorker(s.Language, kind)
if err != nil {
return nil, err
}
if err := ctx.Err(); err != nil {
return nil, err
}
p, err := pull(worker, wait)
if err != nil || p == nil {
return nil, err
}
fate := func(f string) func() error { return func() error { return settle(p.ID, f) } }
return &Pulled{Data: p.Body, Delivered: p.Delivered, Ack: fate(stdio.SettleAck), InProgress: fate(stdio.SettleProgress),
Nak: fate(stdio.SettleNak), Term: fate(stdio.SettleTerm)}, nil
}
+40
View File
@@ -0,0 +1,40 @@
package buildstep
import (
"context"
"testing"
"time"
stdio "git.novox.be/novox/mesh-sdk/go"
"git.novox.be/novox/mesh-sdk/go/delivery"
)
// 1887/A11: a bundle's source pulls the kind's worker by its seat's name and settles each ask by the runtime's id.
func TestSeatAsksPullsTheKindsWorkerAndSettlesByID(t *testing.T) {
var pulled []string
var settled []string
src := SeatAsks{Language: delivery.AskGo,
Pull: func(worker string, wait time.Duration) (*stdio.Pulled, error) {
pulled = append(pulled, worker)
if worker != "SEAT_NODE_BUILD_GO_CHECK_worker" {
return nil, nil
}
return &stdio.Pulled{ID: "p9", Worker: worker, Body: []byte(`{"v":1}`), Delivered: 2}, nil
},
Settle: func(id, fate string) error { settled = append(settled, id+" "+fate); return nil }}
if p, err := src.Next(context.Background(), delivery.AskDelivery, 0); err != nil || p != nil {
t.Fatalf("delivery = %+v, %v", p, err)
}
p, err := src.Next(context.Background(), delivery.AskCheck, time.Second)
if err != nil || p == nil || string(p.Data) != `{"v":1}` || p.Delivered != 2 {
t.Fatalf("check = %+v, %v", p, err)
}
_ = p.InProgress()
_ = p.Ack()
if len(pulled) != 2 || pulled[0] != "SEAT_NODE_BUILD_GO_DELIVERY_worker" {
t.Errorf("pulled %v", pulled)
}
if len(settled) != 2 || settled[0] != "p9 progress" || settled[1] != "p9 ack" {
t.Errorf("settled %v", settled)
}
}
+265
View File
@@ -0,0 +1,265 @@
package buildstep
import (
"context"
"encoding/json"
"sync"
"testing"
"time"
"git.novox.be/novox/mesh-sdk/go/delivery"
)
// steps records a steps state's puts: the last value of each key.
type steps struct {
mu sync.Mutex
keys map[string]delivery.Step
}
func newSteps() *steps { return &steps{keys: map[string]delivery.Step{}} }
func (s *steps) put(key string, value any) (uint64, error) {
raw, _ := value.(json.RawMessage)
st, err := delivery.DecodeStep(raw)
if err != nil {
return 0, err
}
s.mu.Lock()
s.keys[key] = st
s.mu.Unlock()
return 1, nil
}
func (s *steps) get(key string) (delivery.Step, bool) {
s.mu.Lock()
defer s.mu.Unlock()
st, ok := s.keys[key]
return st, ok
}
func testAsk(kind string) delivery.StepAsk {
id := NewID()
a := delivery.StepAsk{ID: id, Kind: kind, Language: delivery.AskGo, Asker: "mesh-delivery-builder", At: time.Now()}
fetch := map[string]any{"repository": "novox/x", "commit": "0123456789abcdef0123456789abcdef01234567"}
if kind == delivery.AskDelivery {
b := buildJobFixture
b.Fetch = fetch
a.Build = &b
} else {
a.Check = &delivery.CheckJob{Fetch: fetch, Levels: []string{delivery.LevelUnit}}
}
return a
}
var buildJobFixture = delivery.BuildJob{Module: "m", Path: ".", Artifacts: []delivery.AskArtifact{{Artifact: "bundle"}}}
func waitFor(t *testing.T, what string, cond func() bool) {
t.Helper()
deadline := time.Now().Add(2 * time.Second)
for !cond() {
if time.Now().After(deadline) {
t.Fatalf("waited 2 s for %s", what)
}
time.Sleep(time.Millisecond)
}
}
// 1887/A2a: given a check ask published before a delivery ask and one slot, the delivery ask runs first.
func TestPullTakesDeliveryBeforeCheck(t *testing.T) {
src, st := NewFakeAsks(), newSteps()
check, recheck, deliver := testAsk(delivery.AskCheck), testAsk(delivery.AskRecheck), testAsk(delivery.AskDelivery)
for _, a := range []delivery.StepAsk{recheck, check, deliver} {
if err := src.PublishAsk(a); err != nil {
t.Fatal(err)
}
}
var mu sync.Mutex
var order []string
p := &Puller{Source: src, Language: delivery.AskGo, Node: "shanks", Put: st.put, Capacity: func() int { return 1 },
PullWait: 10 * time.Millisecond, Run: func(ctx context.Context, a delivery.StepAsk) (any, error) {
mu.Lock()
order = append(order, a.Kind)
mu.Unlock()
return map[string]string{"ran": a.ID}, nil
}}
ctx, cancel := context.WithCancel(context.Background())
defer cancel()
go p.Pull(ctx)
waitFor(t, "three asks acked", func() bool { a, _, _ := src.Fates(); return len(a) == 3 })
mu.Lock()
defer mu.Unlock()
if want := []string{delivery.AskDelivery, delivery.AskCheck, delivery.AskRecheck}; len(order) != 3 || order[0] != want[0] || order[1] != want[1] || order[2] != want[2] {
t.Fatalf("order = %v, want %v", order, want)
}
key, _ := delivery.AskStepKey("shanks", deliver.ID)
s, ok := st.get(key)
if !ok || s.State != delivery.StepPassed || string(s.Result) != `{"ran":"`+deliver.ID+`"}` {
t.Fatalf("the step %s = %+v (%v)", key, s, ok)
}
}
// 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.
func TestPullHoldsAtMostItsCapacity(t *testing.T) {
src, st := NewFakeAsks(), newSteps()
for i := 0; i < 3; i++ {
_ = src.PublishAsk(testAsk(delivery.AskDelivery))
}
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 },
PullWait: 10 * time.Millisecond, Run: func(ctx context.Context, a delivery.StepAsk) (any, error) {
mu.Lock()
started++
mu.Unlock()
<-release
return nil, nil
}}
ctx, cancel := context.WithCancel(context.Background())
defer cancel()
go p.Pull(ctx)
waitFor(t, "two in hand", func() bool { return p.InHand() == 2 })
time.Sleep(30 * time.Millisecond) // three pull waits: the third must still wait
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))
}
release <- struct{}{}
waitFor(t, "the third taken", func() bool { return src.Waiting(delivery.AskDelivery) == 0 })
mu.Lock()
if started != 3 {
t.Errorf("started %d, want 3", started)
}
mu.Unlock()
close(release)
waitFor(t, "all acked", func() bool { a, _, _ := src.Fates(); return len(a) == 3 })
}
// An ask that is not one, or not this seat's, is dropped and runs nowhere.
func TestPullDropsWhatIsNoAsk(t *testing.T) {
src, st := NewFakeAsks(), newSteps()
src.Publish(delivery.AskDelivery, []byte(`{"v":1}`))
other := testAsk(delivery.AskDelivery)
other.Language = delivery.AskNode
_ = src.PublishAsk(other)
ran := false
p := &Puller{Source: src, Language: delivery.AskGo, Node: "shanks", Put: st.put, Capacity: func() int { return 1 },
PullWait: 10 * time.Millisecond, Run: func(ctx context.Context, a delivery.StepAsk) (any, error) { ran = true; return nil, nil }}
ctx, cancel := context.WithCancel(context.Background())
defer cancel()
go p.Pull(ctx)
waitFor(t, "two termed", func() bool { _, tm, _ := src.Fates(); return len(tm) == 2 })
if ran {
t.Fatal("a wrong ask ran")
}
}
// load reads the recorded steps state's raw values: the pause key is not a step, so it is kept apart.
type kv struct {
mu sync.Mutex
vals map[string][]byte
}
func (k *kv) put(key string, value any) (uint64, error) {
raw, _ := value.(json.RawMessage)
k.mu.Lock()
defer k.mu.Unlock()
if k.vals == nil {
k.vals = map[string][]byte{}
}
k.vals[key] = append([]byte(nil), raw...)
return 1, nil
}
func (k *kv) load(key string) ([]byte, error) {
k.mu.Lock()
defer k.mu.Unlock()
return k.vals[key], nil
}
// 1887/A2b, 1887/AC-zero: at capacity 0 the puller takes nothing; a capacity raised takes effect at the next pull.
func TestPullAtCapacityZeroTakesNothing(t *testing.T) {
src, st := NewFakeAsks(), &kv{}
_ = src.PublishAsk(testAsk(delivery.AskDelivery))
var mu sync.Mutex
capacity := 0
p := &Puller{Source: src, Language: delivery.AskGo, Node: "novox", Put: st.put, PullWait: 10 * time.Millisecond,
Capacity: func() int { mu.Lock(); defer mu.Unlock(); return capacity },
Run: func(ctx context.Context, a delivery.StepAsk) (any, error) { return nil, nil }}
ctx, cancel := context.WithCancel(context.Background())
defer cancel()
go p.Pull(ctx)
time.Sleep(50 * time.Millisecond)
if src.Waiting(delivery.AskDelivery) != 1 {
t.Fatal("a node at capacity 0 took an ask")
}
mu.Lock()
capacity = 1
mu.Unlock()
p.Wake()
waitFor(t, "the ask taken at capacity 1", func() bool { a, _, _ := src.Fates(); return len(a) == 1 })
}
// 1887/A2b: a paused puller takes nothing, its pause is kept on the pause key, and a puller started again on that state
// stays paused until resumed.
func TestPullPauseOutlivesARestart(t *testing.T) {
src, st := NewFakeAsks(), &kv{}
run := func(ctx context.Context, a delivery.StepAsk) (any, error) { return nil, nil }
p := &Puller{Source: src, Language: delivery.AskGo, Node: "shanks", Put: st.put, Load: st.load, PullWait: 10 * time.Millisecond,
Capacity: func() int { return 2 }, Run: run}
if err := p.Pause("the operator"); err != nil {
t.Fatal(err)
}
key, _ := delivery.PauseKey("shanks")
raw, _ := st.load(key)
if v, err := delivery.DecodePause(raw); err != nil || !v.Paused || v.By != "the operator" {
t.Fatalf("the pause key holds %s (%v)", raw, err)
}
// The restart: a new puller on the same state.
again := &Puller{Source: src, Language: delivery.AskGo, Node: "shanks", Put: st.put, Load: st.load, PullWait: 10 * time.Millisecond,
Capacity: func() int { return 2 }, Run: run}
_ = src.PublishAsk(testAsk(delivery.AskDelivery))
ctx, cancel := context.WithCancel(context.Background())
defer cancel()
go again.Pull(ctx)
time.Sleep(50 * time.Millisecond)
if !again.Paused() || src.Waiting(delivery.AskDelivery) != 1 {
t.Fatalf("restarted: paused %v, waiting %d; want paused and 1", again.Paused(), src.Waiting(delivery.AskDelivery))
}
if err := again.Resume("the operator"); err != nil {
t.Fatal(err)
}
waitFor(t, "the ask taken after the resume", func() bool { a, _, _ := src.Fates(); return len(a) == 1 })
}
// 1887/A2c: a long step tells the bus it is still being worked; a holder that stops mid-step neither ends nor
// acknowledges its ask, so the stream gives it to another node.
func TestPullHeartbeatAndStop(t *testing.T) {
src, st := NewFakeAsks(), newSteps()
a := testAsk(delivery.AskDelivery)
_ = src.PublishAsk(a)
running := make(chan struct{})
p := &Puller{Source: src, Language: delivery.AskGo, Node: "shanks", Put: st.put, Capacity: func() int { return 1 },
PullWait: 10 * time.Millisecond, Heartbeat: 5 * time.Millisecond,
Run: func(ctx context.Context, a delivery.StepAsk) (any, error) {
close(running)
<-ctx.Done()
return nil, ctx.Err()
}}
ctx, cancel := context.WithCancel(context.Background())
stopped := make(chan struct{})
go func() { p.Pull(ctx); close(stopped) }()
<-running
waitFor(t, "two heartbeats", func() bool { src.mu.Lock(); defer src.mu.Unlock(); return src.InProgress >= 2 })
cancel()
<-stopped
waitFor(t, "the step's goroutine gone", func() bool { return p.InHand() == 0 })
acked, termed, naked := src.Fates()
if len(acked)+len(termed)+len(naked) != 0 {
t.Fatalf("a stopped holder gave its ask a fate: acked %v termed %v naked %v", acked, termed, naked)
}
key, _ := delivery.AskStepKey("shanks", a.ID)
if s, ok := st.get(key); !ok || s.State != delivery.StepRunning {
t.Fatalf("the step %s = %+v, want left running", key, s)
}
}
+5
View File
@@ -1,5 +1,7 @@
package buildstep
import "fmt"
// Refusal is a step that is not run on purpose, as against one that failed: the toolchain installed is not the
// recipe's, the integration level's postgres major is not the one it is built for (novox/hq ADR 0469 §1, ADR 0470
// §4; issue 1473). A step's run returns one and the runner ends the step refused with its text.
@@ -11,3 +13,6 @@ type Refusal struct {
}
func (r *Refusal) Error() string { return r.Text }
// Refuse is a Refusal with its text formatted.
func Refuse(format string, args ...any) *Refusal { return &Refusal{Text: fmt.Sprintf(format, args...)} }
+40
View File
@@ -0,0 +1,40 @@
package buildstep
import (
stdio "git.novox.be/novox/mesh-sdk/go"
)
// SeatTools are a build seat's verbs on one node, as its holder serves them (novox/hq ADR 0525 §10; mesh-sdk catalogue
// buildSeatVerbs): `<seat>.current` (what the node runs, its capacity, whether it is paused), `<seat>.pause` and
// `<seat>.resume`, the pause kept on the node's pause key by the puller. what names the asks in the descriptions ("Go").
func SeatTools(seat, what string, p *Puller) []stdio.Tool {
verb := func(name, description string, run func() (any, error)) stdio.Tool {
return stdio.Tool{Name: seat + "." + name, Description: description,
Input: map[string]any{"type": "object", "properties": map[string]any{}},
Run: func(map[string]any) (any, error) { return run() }}
}
current := func() (any, error) {
capacity := 0
if p.Capacity != nil {
capacity = p.Capacity()
}
return map[string]any{"node": p.Node, "capacity": capacity, "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, "+
"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) {
if err := p.Pause("the seat's pause verb"); err != nil {
return nil, err
}
return current()
}),
verb("resume", "Take "+what+" asks again on this node.", func() (any, error) {
if err := p.Resume("the seat's resume verb"); err != nil {
return nil, err
}
return current()
}),
}
}
+52
View File
@@ -0,0 +1,52 @@
package buildstep
import (
"context"
"testing"
"time"
"git.novox.be/novox/mesh-sdk/go/delivery"
)
// 1887/A9a, ADR 0525 §10: a build seat's verbs are <seat>.current, .pause and .resume; current names the ask running,
// pause stops the puller taking asks, resume lets it take them again.
func TestSeatToolsCurrentPauseResume(t *testing.T) {
src, st := NewFakeAsks(), newSteps()
release := make(chan struct{})
p := &Puller{Source: src, Language: delivery.AskGo, Node: "shanks", Put: st.put, Capacity: func() int { return 1 },
PullWait: 10 * time.Millisecond, Run: func(ctx context.Context, a delivery.StepAsk) (any, error) { <-release; return nil, nil }}
tools := map[string]func(map[string]any) (any, error){}
for _, tool := range SeatTools("node-build-go", "Go", p) {
tools[tool.Name] = tool.Run
}
for _, name := range []string{"node-build-go.current", "node-build-go.pause", "node-build-go.resume"} {
if tools[name] == nil {
t.Fatalf("no %s", name)
}
}
a := testAsk(delivery.AskDelivery)
_ = src.PublishAsk(a)
ctx, cancel := context.WithCancel(context.Background())
defer cancel()
go p.Pull(ctx)
waitFor(t, "the ask running", func() bool { return p.InHand() == 1 })
out, _ := tools["node-build-go.current"](nil)
cur := out.(map[string]any)
running := cur["running"].([]RunningAsk)
if len(running) != 1 || running[0].ID != a.ID || running[0].Module != "m" || cur["capacity"] != 1 || cur["paused"] != false {
t.Fatalf("current = %+v", cur)
}
if out, err := tools["node-build-go.pause"](nil); err != nil || out.(map[string]any)["paused"] != true {
t.Fatalf("pause = %v, %v", out, err)
}
close(release)
_ = src.PublishAsk(testAsk(delivery.AskDelivery))
time.Sleep(50 * time.Millisecond)
if src.Waiting(delivery.AskDelivery) != 1 {
t.Fatal("a paused node took an ask")
}
if _, err := tools["node-build-go.resume"](nil); err != nil {
t.Fatal(err)
}
waitFor(t, "the ask taken after the resume", func() bool { return src.Waiting(delivery.AskDelivery) == 0 })
}
+48 -31
View File
@@ -42,43 +42,60 @@ func NewID() string {
// the error's text, or refused when the error is a *Refusal; a panic in run ends the step failed, naming it, and
// the process goes on. If the running write fails, nothing was started and Start answers the error.
func Start(ctx context.Context, put Put, node, tool string, run func(ctx context.Context) (result any, err error)) (string, error) {
step := delivery.Step{ID: NewID(), Tool: tool, Node: node, State: delivery.StepRunning, Started: now().UTC()}
key, err := delivery.StepKey(node, step.ID)
step, key, err := begin(put, node, tool, NewID())
if err != nil {
return "", err
}
if err := write(put, key, step); err != nil {
return "", fmt.Errorf("buildstep: writing the running step %s: %w", key, err)
}
go func() {
result, err := protect(ctx, run)
step.Ended = now().UTC()
switch ref := (*Refusal)(nil); {
case err == nil:
step.State = delivery.StepPassed
if result != nil {
raw, merr := json.Marshal(result)
if merr != nil {
step.State, step.Reason = delivery.StepFailed, "the result is not JSON: "+merr.Error()
} else {
step.Result = raw
}
}
case errors.As(err, &ref):
step.State, step.Reason = delivery.StepRefused, err.Error()
default:
step.State, step.Reason = delivery.StepFailed, err.Error()
}
for attempt := 1; ; attempt++ {
if werr := write(put, key, step); werr == nil || attempt >= EndAttempts {
return
}
time.Sleep(EndRetry)
}
}()
go end(ctx, put, key, step, run)
return step.ID, nil
}
// begin writes the step running under id.
func begin(put Put, node, tool, id string) (delivery.Step, string, error) {
step := delivery.Step{ID: id, Tool: tool, Node: node, State: delivery.StepRunning, Started: now().UTC()}
key, err := delivery.StepKey(node, step.ID)
if err != nil {
return delivery.Step{}, "", err
}
if err := write(put, key, step); err != nil {
return delivery.Step{}, "", fmt.Errorf("buildstep: writing the running step %s: %w", key, err)
}
return step, key, nil
}
// end runs run and writes the step's end.
func end(ctx context.Context, put Put, key string, step delivery.Step, run func(ctx context.Context) (any, error)) {
result, err := protect(ctx, run)
finish(put, key, step, result, err)
}
// finish writes the step's end from run's result or error: passed, failed or refused (a *Refusal).
func finish(put Put, key string, step delivery.Step, result any, err error) {
step.Ended = now().UTC()
switch ref := (*Refusal)(nil); {
case err == nil:
step.State = delivery.StepPassed
if result != nil {
raw, merr := json.Marshal(result)
if merr != nil {
step.State, step.Reason = delivery.StepFailed, "the result is not JSON: "+merr.Error()
} else {
step.Result = raw
}
}
case errors.As(err, &ref):
step.State, step.Reason = delivery.StepRefused, err.Error()
default:
step.State, step.Reason = delivery.StepFailed, err.Error()
}
for attempt := 1; ; attempt++ {
if werr := write(put, key, step); werr == nil || attempt >= EndAttempts {
return
}
time.Sleep(EndRetry)
}
}
// protect is run with a panic turned into the error that ends the step.
func protect(ctx context.Context, run func(ctx context.Context) (any, error)) (result any, err error) {
defer func() {
+4 -4
View File
@@ -170,10 +170,10 @@ func TestAMembershipForTheNewBusIsComposedAsASealedFile(t *testing.T) {
func TestAStoreRowWithoutAProtocolKeepsTheCompiledOne(t *testing.T) {
was := Seats()
t.Cleanup(func() { UseSeats(was) })
UseSeats([]Seat{{Name: "mesh-build-machine", Scope: ScopeMesh, Decision: "row"}})
got, ok := SeatNamed("mesh-build-machine")
if !ok || len(got.Accepts) == 0 {
t.Fatalf("the build machine's seat lost what it accepts when loaded from the store: %+v", got)
UseSeats([]Seat{{Name: "node-build-go", Scope: ScopeNode, Decision: "row"}})
got, ok := SeatNamed("node-build-go")
if !ok || len(got.Accepts) == 0 || !got.PulledByKind {
t.Fatalf("the Go build seat lost what it accepts when loaded from the store: %+v", got)
}
if got.Decision != "row" {
t.Fatalf("the store's own columns were not kept: %+v", got)
+41 -38
View File
@@ -45,6 +45,10 @@ type Seat struct {
// today — which the bus refuses, because a namespace belongs to who it is named for.
Accepts []string
Emits []string
// PulledByKind says each accepted kind is a worker of its own on the seat's stream, which its holders pull in the
// order Accepts lists them (novox/hq ADR 0525 §2, ADR 0536 §4): the build seats'. Compiled, never stored, like
// Receives.
PulledByKind bool
// Serves carries each verb in full — name, description, schema — because a role's tools are the
// mesh's to define and an agent's to call (novox/hq ADR 0132, design 33 §2).
Serves []Verb
@@ -238,31 +242,19 @@ var defaultSeats = append([]Seat{
// They keep their names until that migration is done deliberately, apart from the node-* pass.
{Name: "npm-package-registry", Scope: ScopeMesh, Delivers: "npm-package-registry", Decision: "novox/hq ADR 0109"},
{Name: "git", Scope: ScopeMesh, Delivers: "git", Decision: "novox/hq ADR 0111"},
// A build is work submitted to this role and its outcome is the role's own event (ADR 0129).
// One publish reaches whoever asked, the controller that records it, and the catalogue that
// places it in the graph — what the old bus's shared exchange did for free.
// A build says what it does as it does it (novox/hq ADR 0157): `started` when work is taken,
// `log.<build id>` for every line, `built` for the outcome. The log's tail token is the build's
// id, so a reader follows one build by subject alone.
// **Node-scoped, and every holder takes from one queue** (novox/hq ADR 0190): a build is asked of
// the role, and whichever machine holding the seat is idle pulls it. One holder per machine is
// what the scope says; sharing the work is what a seat's queue has always done.
//
// **And its holder answers for the build it is running** (novox/hq ADR 0219): what it is
// building, kill it, take nothing new, take again. The queue as a whole is the controller's to
// show and change (`queue`, `cancel`, `clear`); what one machine does with an ask it already
// took only that machine can do. `paused.<node>` is each holder saying whether it takes work, so
// the controller can tell a plan waiting on a paused seat from one that is late.
{Name: "node-build-agent", Scope: ScopeNode,
Accepts: []string{"build"}, Emits: []string{"started", "built", "log.*", "paused.*"},
Serves: buildAgentVerbs(),
Decision: "novox/hq ADR 0190, ADR 0219"},
// **Retired by ADR 0190, kept while a manifest still claims it.** The one build machine's seat.
// A claim to a seat the mesh no longer defines is refused, and the module holding this one is
// assigned on a live machine until build-agent replaces it — removing the row first would make
// that machine unresolvable in the meantime. Deleted once no registered manifest claims it.
{Name: "mesh-build-machine", Scope: ScopeMesh,
Accepts: []string{"build"}, Emits: []string{"started", "built", "log.*"}, Decision: "novox/hq ADR 0190"},
// **A build node's work, one seat per build language** (novox/hq ADR 0525 §2, §10; ADR 0536 §4; issue 1887). The
// build module of each language claims its seat on every node it runs on; the builder and the checker put their
// asks on the seat's accepted kinds, which the broker derives as one work queue with a shared worker per kind, and
// each holder pulls delivery first, then check, then recheck, at most its node's build capacity at once. Its verbs
// say what the node runs and pause or resume it there: every holder pulls from the same worker, so a node's pause is
// its holder taking nothing, kept on the module's own pause key (issue 1887's design limit 2). They replace
// node-build-agent, retired with build-agent (ADR 0524 §5).
{Name: "node-build-go", Scope: ScopeNode, Accepts: buildAskKinds(), PulledByKind: true, Serves: buildSeatVerbs("Go"),
Decision: "novox/hq ADR 0525, ADR 0536"},
{Name: "node-build-node", Scope: ScopeNode, Accepts: buildAskKinds(), PulledByKind: true, Serves: buildSeatVerbs("Node"),
Decision: "novox/hq ADR 0525, ADR 0536"},
{Name: "node-build-image", Scope: ScopeNode, Accepts: buildAskKinds(), PulledByKind: true, Serves: buildSeatVerbs("image"),
Decision: "novox/hq ADR 0525, ADR 0536"},
// **The mesh's resolvers** (novox/hq ADR 0194, 0196, 0223): every node's internal domain, and
// every node and container asks them and nothing else. Replicated since ADR 0223: held on more
// than one machine, each answering the same names from the same roster, and every machine's
@@ -498,6 +490,7 @@ func UseSeats(s []Seat) {
row.Receives = d.Receives
row.Paths = d.Paths
row.Replicated = d.Replicated
row.PulledByKind = d.PulledByKind
row.Serves = optionalAsCompiled(row.Serves, d.Serves)
}
merged = append(merged, row)
@@ -623,6 +616,10 @@ func claimProblems(m Manifest) []string {
// after UseSeats has replaced the set with the store's.
continue
}
if replaced, retired := RetiredSeats[c.Name]; retired {
problems = append(problems, fmt.Sprintf("%s claims the seat %q, which is retired: its work is %s", m.Module, c.Name, replaced))
continue
}
if isSystemSeatName(c.Name) {
problems = append(problems, fmt.Sprintf(
"%s claims %q, which is a seat in the mesh's own namespace (mesh-*/node-*) that it "+
@@ -903,22 +900,28 @@ func loginShellVerbs() []Verb {
}
}
// buildAgentVerbs are what a holder of node-build-agent answers on its own machine (novox/hq ADR
// 0219). One build at a time per holder (ADR 0190), so "the build running here" is one or none.
func buildAgentVerbs() []Verb {
// buildAskKinds are a build seat's accepted kinds, in the order a holder pulls them (ADR 0536 §4; mesh-sdk
// delivery.AskKinds).
func buildAskKinds() []string { return []string{"delivery", "check", "recheck"} }
// RetiredSeats are seats the mesh no longer defines, each with what replaced it: a claim or a use of one is refused
// naming its replacement, not as a seat nobody declares.
var RetiredSeats = map[string]string{
"node-build-agent": "node-build-go, node-build-node and node-build-image, held by the build modules (novox/hq ADR 0525 §10)",
"mesh-build-machine": "node-build-go, node-build-node and node-build-image, held by the build modules (novox/hq ADR 0190, ADR 0525 §10)",
}
// buildSeatVerbs are what a build module answers on its own node for its seat (ADR 0525 §10).
func buildSeatVerbs(language string) []Verb {
return []Verb{
{Name: "current", Description: "The build this machine is running — its id, repository, path, " +
"the step it is at, when it started and for how long — or none; and whether this machine is " +
"paused, taking no new build.",
{Name: "current", Description: "The " + language + " asks this node runs now (each ask's id, kind, module and when it " +
"started), its build capacity, and whether it is paused, taking no new ask.",
Input: schema(map[string]string{}, nil)},
{Name: "kill", Description: "End the build with this id, running here: its commands and the " +
"containers it started are stopped, and its outcome is announced as failed, killed by hand — " +
"settled, so it is not handed to another machine.",
Input: schema(map[string]string{"id": "the build's id, as `queue` or `current` says it"}, []string{"id"})},
{Name: "pause", Description: "Take no new build on this machine until resumed; a build running " +
"here finishes. Kept across a restart of the holder.",
{Name: "pause", Description: "Take no new " + language + " ask on this node until resumed; an ask running here " +
"ends in its place. Kept across a restart of the build module (its pause key). A capacity of 0 is the " +
"lasting way to keep a node from building.",
Input: schema(map[string]string{}, nil)},
{Name: "resume", Description: "Take builds again on this machine.",
{Name: "resume", Description: "Take " + language + " asks again on this node.",
Input: schema(map[string]string{}, nil)},
}
}
+8
View File
@@ -364,6 +364,10 @@ func CatalogueProblems(shelf Shelf) []string {
// A `uses` naming nothing is where ADR 0110's guarantee lands under a derived set: the
// same refusal, at the same moment, from a set nobody maintains by hand.
for _, u := range m.Uses {
if replaced, retired := RetiredSeats[u]; retired && !exists(u) {
problems = append(problems, fmt.Sprintf("%s uses the seat %q, which is retired: its work is %s", module, u, replaced))
continue
}
if !exists(u) {
problems = append(problems, fmt.Sprintf(
"%s uses the seat %q, "+UndeclaredSeat,
@@ -372,6 +376,10 @@ func CatalogueProblems(shelf Shelf) []string {
}
for _, c := range m.Claims {
if replaced, retired := RetiredSeats[c.Name]; retired && !exists(c.Name) {
problems = append(problems, fmt.Sprintf("%s claims the seat %q, which is retired: its work is %s", module, c.Name, replaced))
continue
}
if !exists(c.Name) {
problems = append(problems, fmt.Sprintf(
"%s claims the seat %q, "+UndeclaredSeat,
+36 -3
View File
@@ -43,7 +43,8 @@ func TestTheSeatsAreAClosedSetAndEachNamesItsDecision(t *testing.T) {
delivered[s.Delivers] = s.Name
}
}
// Forty-five with node-files (novox/hq ADR 0394); forty-four again once node-reads goes.
// Forty-six with the three build seats replacing node-build-agent and the retired mesh-build-machine (novox/hq
// ADR 0525 §2, §10); forty-five with node-files (novox/hq ADR 0394); forty-four again once node-reads goes.
// Forty-four with node-provisioner (novox/hq ADR 0349 clause 3); forty-three with mesh-provisioner (novox/hq ADR 0349 clause 3); forty-two with mesh-config (novox/hq ADR 0348 §9); forty-one with node-reads (novox/hq ADR 0310); forty with node-nfs-server and node-mounts (novox/hq ADR 0263); thirty-eight with node-resolver (novox/hq ADR 0247); thirty-seven with mesh-delivery (novox/hq ADR 0239); thirty-six since node-resolver-config retired
// into node-uplink (novox/hq ADR 0223); thirty-seven
// since the retired node-dns-resolver went (novox/hq ADR 0220); thirty-eight with
@@ -55,8 +56,8 @@ func TestTheSeatsAreAClosedSetAndEachNamesItsDecision(t *testing.T) {
// node-container-runtime (ADR 0207); nineteen with node-environment and node-login-shell (ADR 0203,
// ADR 0204); seventeen with node-build-agent (ADR 0190). One fewer once the retired
// mesh-build-machine row goes, when no registered manifest claims it.
if len(Seats()) != 45 {
t.Errorf("the mesh defines %d seats rather than 45; the set is closed, so a change here is "+
if len(Seats()) != 46 {
t.Errorf("the mesh defines %d seats rather than 46; the set is closed, so a change here is "+
"a decision (novox/hq ADR 0110): %s", len(Seats()), seatNames())
}
}
@@ -300,3 +301,35 @@ func TestAFormerNameResolvesAfterARename(t *testing.T) {
t.Fatalf("the holder recorded under the former name was not matched: %+v found=%v", holder, found)
}
}
// 1887/A9a: the catalogue defines no seat node-build-agent, and a manifest claiming it is refused naming its
// replacement; the build modules' seats accept the three kinds in their pull order and serve current, pause, resume.
func TestNodeBuildAgentIsRetiredForTheBuildSeats(t *testing.T) {
for _, gone := range []string{"node-build-agent", "mesh-build-machine"} {
if _, ok := SeatNamed(gone); ok {
t.Errorf("the mesh still defines %s", gone)
}
}
_, err := ParseManifest([]byte(`{"module":"build-agent","version":"1","claims":[{"name":"node-build-agent","scope":"node"}]}`))
if err == nil || !strings.Contains(err.Error(), `"node-build-agent", which is retired`) || !strings.Contains(err.Error(), "node-build-go") {
t.Fatalf("the claim is not refused naming its replacement: %v", err)
}
// And over a catalogue whose store still holds no row for it, a use of it is refused the same way.
user := Manifest{Module: "watcher", Uses: []string{"node-build-agent"}}
if joined := strings.Join(CatalogueProblems(Shelf{user.Module: user}), "; "); !strings.Contains(joined, "which is retired: its work is node-build-go") {
t.Fatalf("a use is not refused naming the replacement: %s", joined)
}
for _, name := range []string{"node-build-go", "node-build-node", "node-build-image"} {
s, ok := SeatNamed(name)
if !ok || s.Scope != ScopeNode || !s.PulledByKind || strings.Join(s.Accepts, ",") != "delivery,check,recheck" {
t.Fatalf("%s = %+v", name, s)
}
var verbs []string
for _, v := range s.Serves {
verbs = append(verbs, v.Name)
}
if strings.Join(verbs, ",") != "current,pause,resume" {
t.Errorf("%s serves %v", name, verbs)
}
}
}
+12 -33
View File
@@ -350,36 +350,24 @@ var ControllerVerbs = []Verb{
"its shell — `node show ace`, `module list`, `plans`, `conditions show <key>` — and answer what it " +
"printed. The generic verb beside the named ones (novox/hq ADR 0154), and since ADR 0266 it only reads: " +
"status, version, help, seats, healers, hand-acts, durations, collection, images, artifacts, data, builds, " +
"queue, plan, plans (not stop/close/go), doctor (not run), conditions list/show/history, node list/show, " +
"plan, plans (not stop/close/go), doctor (not run), conditions list/show/history, node list/show, " +
"module list, settings show/preferences, retire list, cleanup list, delivery plan/walks, bus, mirrors (not " +
"--record/--confirm). What writes has its named verb; what sets a key, issues a credential or a token, or " +
"accepts, recovers or exports a secret is the controller's terminal's alone.",
Input: schema(map[string]string{
"command": "the command line, as the controller's binary takes it; quotes group a word with spaces",
}, []string{"command"})},
// The build queue, controlled by hand (novox/hq ADR 0219). Every verb that drops an ask leaves a
// failed outcome for it, so a plan waiting on it fails visibly instead of hanging.
{Name: "queue", Description: "Every ask in the build queue: waiting, in flight (on which machine, for how long), " +
"and dead (handed out as often as allowed and never settled) — each with its id, repository, path, ref and when it was asked. " +
"Answers the table as text and, as data, each ask's state, kind (check or build), repository, commit, machine, since " +
"and position among the waiting (novox/hq issue 455).",
// The build queue went with build-agent (novox/hq ADR 0525 §9): asks are the build seats' streams, their backlog the
// streams' own state, shown in the delivery view. The verbs are refused by name for one release, naming that.
{Name: "queue", Description: "Removed in R4g (ADR 0525 §9): refused by name, naming its replacement: the build seats' backlog in the delivery view",
Input: schema(nil, nil)},
{Name: "cancel", Description: "Drop one waiting or dead ask from the build queue; its outcome is recorded failed, " +
"cancelled by hand, and a plan that asked for it fails. One in flight is refused: `kill` ends it where it runs.",
Input: schema(map[string]string{"id": "the ask's build id, as `queue` lists it"}, []string{"id"})},
{Name: "clear", Description: "Cancel every waiting ask in the build queue — and with dead, every dead one too — each " +
"recorded failed, cancelled by hand. Never touches one in flight.",
Input: schema(map[string]string{"dead": "\"true\" to cancel the dead asks as well"}, nil, "dead")},
{Name: "rebuild", Description: "Removed in R4e (ADR 0501 §5 / ADR 0503 §3): refused by name, naming its replacement",
Input: schema(nil, nil)},
{Name: "replay", Description: "Ask a recorded build's repository and path again at the commit it built, under a new id: " +
"a dry run only, nothing recorded or registered (novox/hq ADR 0429 §18: a fingerprint's key is built again by the " +
"builder, never registered by hand).",
Input: schema(map[string]string{"id": "the build's id"}, []string{"id"})},
{Name: "cancel", Description: "Removed in R4g (ADR 0525 §9): refused by name, naming its replacement", Input: schema(nil, nil)},
{Name: "clear", Description: "Removed in R4g (ADR 0525 §9): refused by name, naming its replacement", Input: schema(nil, nil)},
{Name: "replay", Description: "Removed in R4g (ADR 0525 §9): refused by name, naming its replacement", Input: schema(nil, nil)},
// novox/hq issue 686: a builder ran a branch's store tests for the first time in its pull request's merge check.
{Name: "try", Description: "Try a pushed branch or commit before its pull request (novox/hq issue 686): the " +
"checker (mesh-delivery-checker) merges the head onto the repository's trunk and runs its levels on the merge, " +
"as it judges a pull request (ADR 0503 §3); no merge-check.sh is run and build-agent is not asked. Answers at " +
"as it judges a pull request (ADR 0503 §3), its levels asks on the build seats (ADR 0525 §1); no merge-check.sh is run. Answers at " +
"once with an id; asked again with that id it waits up to 20 s and answers pass, fail or error, how long it " +
"took and the lines naming what failed — or that it is still waiting or running, to ask again. Sets no " +
"status on the forge, registers and builds nothing for delivery. One try of a commit at a time — a second " +
@@ -388,23 +376,14 @@ var ControllerVerbs = []Verb{
"repository": "the repository on the forge, owner/name; with ref",
"ref": "the branch or commit to try, as pushed to the forge",
"base": "the branch it is merged into (the repository's default branch when none is named)",
"tests": "only the tests this `go test -run` pattern names, where the repository's merge-check.sh reads " +
"MESH_CHECK_TESTS; a script that does not runs whole",
"tests": "only the tests this `go test -run` pattern names",
"store-only": "\"true\": only the integration level",
"id": "instead of a repository: the id a try answered, to read what came of it",
}, nil, "store-only"),
Replaces: []string{"sh merge-check.sh", "docker run postgres"}},
// novox/hq ADR 0350 §5: asked by the delivery's owner when the forge says a pull request closed unmerged.
{Name: "forget-checks", Description: "Removed in R4e (ADR 0501 §5 / ADR 0503 §3): refused by name, naming its replacement",
Input: schema(nil, nil)},
{Name: "kill", Description: "End a build where it runs: the machine that took it stops its commands and containers and " +
"announces it failed, killed by hand — settled, never handed to another machine.",
Input: schema(map[string]string{"id": "the build's id"}, []string{"id"})},
{Name: "pause", Description: "The build seat's holder on one machine — or every holder — takes no new build until resumed; " +
"a build running finishes. Kept across a restart of the holder. A plan waiting on a paused seat says so and is not late.",
Input: schema(map[string]string{"node": "one machine; every holder when absent"}, nil)},
{Name: "resume", Description: "The build seat's holder on one machine — or every holder — takes builds again.",
Input: schema(map[string]string{"node": "one machine; every holder when absent"}, nil)},
{Name: "kill", Description: "Removed in R4g (ADR 0525 §9): refused by name, naming its replacement: a build seat's ask ends where it runs", Input: schema(nil, nil)},
{Name: "pause", Description: "Removed in R4g (ADR 0525 §9): refused by name, naming its replacement: <node>/node-build-<language>.pause", Input: schema(nil, nil)},
{Name: "resume", Description: "Removed in R4g (ADR 0525 §9): refused by name, naming its replacement: <node>/node-build-<language>.resume", Input: schema(nil, nil)},
// Acts done by hand, and what the bounds are set from (novox/hq to-be 45 §7, Phase 0).
{Name: "hand-act", Description: "Record an act done by hand outside the mesh — a container restarted, a file " +
"edited, a service started on a machine — with why and its cause, in the hand-act log beside the pushes and " +
+31
View File
@@ -0,0 +1,31 @@
package catalogue
import (
"strings"
"testing"
)
// 1887/A9b: the build queue's verbs are refused by name, naming what replaced them, and the verbs R4e removed are gone.
func TestTheBuildQueueVerbsAreRefusedByName(t *testing.T) {
rows := map[string]Verb{}
for _, v := range ControllerVerbs {
rows[v.Name] = v
}
for _, gone := range []string{"rebuild", "forget-checks"} {
if _, ok := rows[gone]; ok {
t.Errorf("the controller still lists %s", gone)
}
}
for _, refused := range []string{"queue", "cancel", "clear", "kill", "pause", "resume", "replay"} {
v, ok := rows[refused]
if !ok || !strings.HasPrefix(v.Description, "Removed in R4g (ADR 0525 §9): refused by name") {
t.Errorf("%s = %q, want refused by name", refused, v.Description)
}
}
if !strings.Contains(rows["queue"].Description, "delivery view") || !strings.Contains(rows["pause"].Description, "node-build-") {
t.Errorf("queue and pause do not name their replacements: %q, %q", rows["queue"].Description, rows["pause"].Description)
}
if strings.Contains(rows["try"].Description, "build-agent") {
t.Errorf("try still names build-agent: %q", rows["try"].Description)
}
}
+15
View File
@@ -184,6 +184,10 @@ type BuilderBuild struct {
// does not yet hold: the key is not claimed until that package's trunk key is done and published (novox/hq ADR 0500
// clause 6). A reader that does not know the field reads the key as queued.
Waits string `json:"waits,omitempty"`
// Asks are, on a building key, the asks the builder put for it by build seat language (AskGo, AskNode, AskImage), each
// its id (novox/hq ADR 0525 §1; issue 1887 A4b): a builder that takes the key over at its start waits for those asks'
// steps and never puts them again. A reader that does not know the field reads the key as before.
Asks map[string]string `json:"asks,omitempty"`
}
func (b BuilderBuild) check() error {
@@ -216,6 +220,17 @@ func (b BuilderBuild) check() error {
return fmt.Errorf("delivery: the reference %q of the %s %q %s", a.Reference, a.Kind, a.Name, p)
}
}
for language, id := range b.Asks {
if b.State != BuildBuilding {
return fmt.Errorf("delivery: a %s build names the ask %s; only a building key does", b.State, id)
}
if err := checkAskLanguage(language); err != nil {
return err
}
if err := checkAskID(id); err != nil {
return err
}
}
if b.Waits != "" {
if b.State != BuildQueued {
return fmt.Errorf("delivery: a %s build waits for %q; only a queued key waits for a package", b.State, b.Waits)
+220
View File
@@ -0,0 +1,220 @@
package buildrun
import (
"context"
"errors"
"fmt"
"sync"
"time"
"github.com/nats-io/nats.go"
"github.com/nats-io/nats.go/jetstream"
"git.novox.be/novox/mesh-sdk/go/delivery"
)
// **An asker puts an ask on its build seat's stream and waits for its step** (novox/hq ADR 0525 §1, §6; issue 1887 A4,
// A5). It chooses no node: whichever build node pulls the ask writes `<node>.<ask id>` in its build module's steps state,
// and the asker reads the first such key that is not running. The builder and the checker share it.
// AskPublisher puts an encoded ask on subject, de-duplicated by its id.
type AskPublisher interface {
PublishAsk(ctx context.Context, subject, id string, data []byte) error
}
// AskWaiter waits for an ask's step: the first key `*.<ask id>` of the build module's steps state whose state is not
// running. A watch, not a poll; it returns when the step ends or ctx does.
type AskWaiter interface {
WaitAsk(ctx context.Context, module, askID string) (delivery.Step, error)
}
// Asker puts asks and waits for their steps.
type Asker struct {
Publisher AskPublisher
Waiter AskWaiter
// Bound is how long the asker waits for an ask's step before it stops waiting; zero waits as long as ctx.
Bound time.Duration
}
// Put puts a on its stream: the seat's accept subject of its language and kind.
func (k Asker) Put(ctx context.Context, a delivery.StepAsk) error {
raw, err := delivery.EncodeStepAsk(a)
if err != nil {
return err
}
subject, err := delivery.AskSubject(a.Language, a.Kind)
if err != nil {
return err
}
if err := k.Publisher.PublishAsk(ctx, subject, a.ID, raw); err != nil {
return fmt.Errorf("putting the %s ask %s: %w", a.Kind, a.ID, err)
}
return nil
}
// Wait waits for the step of an ask already put (a builder that started again takes over its own asks with it, never
// putting them twice) and answers it when it ended, whatever its state.
func (k Asker) Wait(ctx context.Context, a delivery.StepAsk) (delivery.Step, error) {
module, err := delivery.BuildModuleOf(a.Language)
if err != nil {
return delivery.Step{}, err
}
wait, cancel := ctx, context.CancelFunc(func() {})
if k.Bound > 0 {
wait, cancel = context.WithTimeoutCause(ctx, k.Bound, fmt.Errorf("the ask %s's step did not end within %s", a.ID, k.Bound))
}
defer cancel()
step, err := k.Waiter.WaitAsk(wait, module, a.ID)
if err != nil {
if cause := context.Cause(wait); cause != nil && ctx.Err() == nil {
err = cause
}
return delivery.Step{}, err
}
return step, nil
}
// 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 {
return delivery.Step{}, err
}
return k.Passed(k.Wait(ctx, a))
}
// Passed is step when it passed, else an error naming where it ended and why.
func (k Asker) Passed(step delivery.Step, err error) (delivery.Step, error) {
if err != nil {
return step, err
}
if step.State != delivery.StepPassed {
return step, fmt.Errorf("the ask %s on %s %s: %s", step.ID, step.Node, step.State, step.Reason)
}
return step, nil
}
// JetStreamAsks is the bus side of an Asker: a JetStream publish de-duplicated by the ask's id (Nats-Msg-Id), and a watch
// of `*.<ask id>` in the build module's steps bucket.
type JetStreamAsks struct {
JS jetstream.JetStream
}
// PublishAsk is AskPublisher's.
func (s JetStreamAsks) PublishAsk(ctx context.Context, subject, id string, data []byte) error {
msg := &nats.Msg{Subject: subject, Data: data, Header: nats.Header{}}
msg.Header.Set(jetstream.MsgIDHeader, id)
_, err := s.JS.PublishMsg(ctx, msg)
return err
}
// WaitAsk is AskWaiter's: the watch hands over what the keys hold now, so a step that ended before it began is read at
// once, and every change after.
func (s JetStreamAsks) WaitAsk(ctx context.Context, module, askID string) (delivery.Step, error) {
pattern, err := delivery.AskStepWatch(askID)
if err != nil {
return delivery.Step{}, err
}
kv, err := s.JS.KeyValue(ctx, delivery.StepsBucket(module))
if err != nil {
return delivery.Step{}, fmt.Errorf("the bucket %s: %w", delivery.StepsBucket(module), err)
}
w, err := kv.Watch(ctx, pattern)
if err != nil {
return delivery.Step{}, err
}
defer w.Stop()
for {
select {
case <-ctx.Done():
return delivery.Step{}, ctx.Err()
case e, ok := <-w.Updates():
if !ok {
return delivery.Step{}, errors.New("the watch of the ask's step ended")
}
if e == nil || e.Operation() != jetstream.KeyValuePut {
continue
}
step, err := delivery.DecodeStep(e.Value())
if err != nil {
continue // not a step (a node's pause key never matches an ask id; anything else is not this ask's)
}
if step.State != delivery.StepRunning {
return step, nil
}
}
}
}
// FakeAsks is a test double of the bus for an asker's tests: each ask put is kept and, when Answer is set, its step is
// ended at once with what Answer says a build node wrote. Safe for concurrent use.
type FakeAsks struct {
Answer func(a delivery.StepAsk) delivery.Step
mu sync.Mutex
asks []delivery.StepAsk
steps map[string]delivery.Step
waits map[string]chan struct{}
}
// PublishAsk is AskPublisher's.
func (f *FakeAsks) PublishAsk(_ context.Context, _, id string, data []byte) error {
a, err := delivery.DecodeStepAsk(data)
if err != nil {
return err
}
f.mu.Lock()
f.asks = append(f.asks, a)
f.mu.Unlock()
if f.Answer != nil {
f.End(f.Answer(a))
}
return nil
}
// End writes an ask's step end, as a build node does.
func (f *FakeAsks) End(step delivery.Step) {
f.mu.Lock()
if f.steps == nil {
f.steps, f.waits = map[string]delivery.Step{}, map[string]chan struct{}{}
}
f.steps[step.ID] = step
ch := f.waits[step.ID]
delete(f.waits, step.ID)
f.mu.Unlock()
if ch != nil {
close(ch)
}
}
// WaitAsk is AskWaiter's.
func (f *FakeAsks) WaitAsk(ctx context.Context, _ string, askID string) (delivery.Step, error) {
f.mu.Lock()
if f.steps == nil {
f.steps, f.waits = map[string]delivery.Step{}, map[string]chan struct{}{}
}
if s, ok := f.steps[askID]; ok {
f.mu.Unlock()
return s, nil
}
ch := f.waits[askID]
if ch == nil {
ch = make(chan struct{})
f.waits[askID] = ch
}
f.mu.Unlock()
select {
case <-ctx.Done():
return delivery.Step{}, ctx.Err()
case <-ch:
}
f.mu.Lock()
defer f.mu.Unlock()
return f.steps[askID], nil
}
// Asks are the asks put so far.
func (f *FakeAsks) Asks() []delivery.StepAsk {
f.mu.Lock()
defer f.mu.Unlock()
return append([]delivery.StepAsk(nil), f.asks...)
}
@@ -0,0 +1,61 @@
//go:build integration
package buildrun
import (
"context"
"encoding/json"
"testing"
"time"
"github.com/nats-io/nats.go"
"github.com/nats-io/nats.go/jetstream"
"git.novox.be/novox/mesh-sdk/go/delivery"
"git.novox.be/novox/mesh-sdk/go/testbus"
)
// 1887/A4a, 1887/AC3 on the real bus: an ask put twice under one id is one message, and the asker reads the step a node
// wrote under `<node>.<ask id>`.
func TestAnAskIsPutOnceAndItsStepIsRead(t *testing.T) {
nc, err := nats.Connect(testbus.URL(t))
if err != nil {
t.Fatal(err)
}
defer nc.Close()
js, _ := jetstream.New(nc)
ctx := context.Background()
if _, err := js.CreateStream(ctx, jetstream.StreamConfig{Name: "SEAT_NODE_BUILD_GO", Subjects: []string{"mesh.seat.node-build-go.accept.>"},
Retention: jetstream.WorkQueuePolicy}); err != nil {
t.Fatal(err)
}
kv, err := js.CreateKeyValue(ctx, jetstream.KeyValueConfig{Bucket: delivery.StepsBucket("mesh-build-go")})
if err != nil {
t.Fatal(err)
}
bus := JetStreamAsks{JS: js}
k := Asker{Publisher: bus, Waiter: bus, Bound: 5 * time.Second}
a := goAsk("mvl0frv18d1960")
for i := 0; i < 2; i++ {
if err := k.Put(ctx, a); err != nil {
t.Fatal(err)
}
}
s, _ := js.Stream(ctx, "SEAT_NODE_BUILD_GO")
if info, _ := s.Info(ctx); info.State.Msgs != 1 {
t.Fatalf("%d messages for one ask put twice", info.State.Msgs)
}
go func() {
time.Sleep(100 * time.Millisecond)
key, _ := delivery.AskStepKey("ace", a.ID)
running, _ := delivery.EncodeStep(delivery.Step{ID: a.ID, Tool: "delivery ask", Node: "ace", State: delivery.StepRunning, Started: time.Now()})
_, _ = kv.Put(ctx, key, running)
passed, _ := delivery.EncodeStep(delivery.Step{ID: a.ID, Tool: "delivery ask", Node: "ace", State: delivery.StepPassed, Started: time.Now(),
Result: json.RawMessage(`{"artifacts":[]}`)})
_, _ = kv.Put(ctx, key, passed)
}()
step, err := k.Passed(k.Wait(ctx, a))
if err != nil || step.Node != "ace" {
t.Fatalf("step %+v, %v", step, err)
}
}
+43
View File
@@ -0,0 +1,43 @@
package buildrun
import (
"context"
"strings"
"testing"
"time"
"git.novox.be/novox/mesh-sdk/go/delivery"
)
func goAsk(id string) delivery.StepAsk {
return delivery.StepAsk{ID: id, Kind: delivery.AskDelivery, Language: delivery.AskGo, Asker: "mesh-delivery-builder", At: time.Now(),
Build: &delivery.BuildJob{Module: "m", Path: ".", Fetch: map[string]any{"repository": "novox/x", "commit": strings.Repeat("a", 40)},
Artifacts: []delivery.AskArtifact{{Artifact: "program"}}}}
}
// 1887/A4a: an asker puts the ask once and answers the step a build node wrote for it, naming the node; one that did not
// pass is an error naming the node and why; a wait of an ask already put puts nothing.
func TestAnAskerPutsOnceAndReadsTheStep(t *testing.T) {
f := &FakeAsks{Answer: func(a delivery.StepAsk) delivery.Step {
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()))
}
f.Answer = func(a delivery.StepAsk) delivery.Step {
return delivery.Step{ID: a.ID, Node: "shanks", State: delivery.StepRefused, Reason: "go 1.27 installed"}
}
if _, err := k.Ask(context.Background(), goAsk("mvl0frv18d1958")); err == nil || !strings.Contains(err.Error(), "shanks refused: go 1.27 installed") {
t.Fatalf("a refused step: %v", err)
}
g := &FakeAsks{}
k = Asker{Publisher: g, Waiter: g, Bound: 20 * time.Millisecond}
if _, err := k.Wait(context.Background(), goAsk("mvl0frv18d1959")); err == nil || !strings.Contains(err.Error(), "did not end within") {
t.Fatalf("an ask whose step never ends: %v", err)
}
if len(g.Asks()) != 0 {
t.Fatal("a wait put an ask")
}
}
+427
View File
@@ -0,0 +1,427 @@
package delivery
import (
"encoding/json"
"fmt"
"regexp"
"strings"
"time"
)
// **Asks: a build or a check step is a message a build node pulls when it has a free slot** (novox/hq ADR 0525, as
// ADR 0536 §4 changes it; issue 1887). The builder puts a trunk fingerprint's build on the delivery stream of the
// artifact's language and a candidate's on the check stream; the checker puts a check of one level on the check stream,
// and a re-judge after a trunk move on the recheck stream. Each build module claims the node seat of its language
// (`node-build-go`, `node-build-node`, `node-build-image`), whose accepted kinds the broker derives as the work queue
// `SEAT_NODE_BUILD_<LANGUAGE>` with one shared pull worker per kind; a holder pulls delivery first, then check, then
// recheck, at most its node's build capacity at once (ADR 0525 §4, §5).
//
// **An ask is one whole job on one node** (ADR 0525 §3): a build ask is the fetch, the build (or copy) of each artifact
// and the push; a check ask is the fetch, the digest and the tests of one level. Its end is written on the build module's
// step key `<node>.<ask id>` (StepsBucket), which the asker watches with AskStepWatch (ADR 0525 §6).
// The kinds of ask: the seat's accepted kinds, in the order a holder pulls them.
const (
AskDelivery = "delivery"
AskCheck = "check"
AskRecheck = "recheck"
)
// AskKinds are the accepted kinds of a build seat, in the order a holder pulls them (ADR 0536 §4).
var AskKinds = []string{AskDelivery, AskCheck, AskRecheck}
// The languages of the build seats: each is one build module's (mesh-build-<language>) and one seat's
// (node-build-<language>).
const (
AskGo = "go"
AskNode = "node"
AskImage = "image"
)
// AskLanguages are the build seats' languages.
var AskLanguages = []string{AskGo, AskNode, AskImage}
// StepAskVersion is the version of an Ask value; a reader refuses another (ErrUnknownVersion).
const StepAskVersion = 1
func checkAskKind(kind string) error {
for _, k := range AskKinds {
if kind == k {
return nil
}
}
return fmt.Errorf("delivery: the ask kind %q is not delivery, check or recheck", kind)
}
func checkAskLanguage(language string) error {
for _, l := range AskLanguages {
if language == l {
return nil
}
}
return fmt.Errorf("delivery: the ask language %q is not go, node or image", language)
}
// BuildSeat is the node seat a build module of language claims: node-build-<language>.
func BuildSeat(language string) (string, error) {
if err := checkAskLanguage(language); err != nil {
return "", err
}
return "node-build-" + language, nil
}
// BuildModuleOf is the build module that holds language's seat: mesh-build-<language>.
func BuildModuleOf(language string) (string, error) {
if err := checkAskLanguage(language); err != nil {
return "", err
}
return "mesh-build-" + language, nil
}
// AskSubject is the subject an ask of kind for language is published on: the seat's accept subject,
// `mesh.seat.node-build-<language>.accept.<kind>` (mesh-controller internal/broker SeatStreams).
func AskSubject(language, kind string) (string, error) {
seat, err := BuildSeat(language)
if err != nil {
return "", err
}
if err := checkAskKind(kind); err != nil {
return "", err
}
return "mesh.seat." + seat + ".accept." + kind, nil
}
// ParseAskSubject is the inverse of AskSubject; it refuses a subject of another seat or kind.
func ParseAskSubject(subject string) (language, kind string, err error) {
rest, ok := strings.CutPrefix(subject, "mesh.seat.node-build-")
if !ok {
return "", "", fmt.Errorf("delivery: %q is not a build seat's accept subject", subject)
}
language, kind, ok = strings.Cut(rest, ".accept.")
if !ok {
return "", "", fmt.Errorf("delivery: %q is not a build seat's accept subject", subject)
}
if want, err := AskSubject(language, kind); err != nil || want != subject {
return "", "", fmt.Errorf("delivery: %q is not a build seat's accept subject: %v", subject, err)
}
return language, kind, nil
}
// AskStream is the work-queue stream of language's seat, as the broker names a seat's stream: SEAT_NODE_BUILD_<LANGUAGE>.
func AskStream(language string) (string, error) {
seat, err := BuildSeat(language)
if err != nil {
return "", err
}
return "SEAT_" + strings.ToUpper(strings.ReplaceAll(seat, "-", "_")), nil
}
// AskWorker is the shared pull consumer of one kind on language's stream, filtered on that kind's subject:
// SEAT_NODE_BUILD_<LANGUAGE>_<KIND>_worker. Every holder of the seat pulls from it, so the work is shared without a
// holder knowing the others (ADR 0190), and the kinds' workers are apart so a holder can pull them in their order.
func AskWorker(language, kind string) (string, error) {
stream, err := AskStream(language)
if err != nil {
return "", err
}
if err := checkAskKind(kind); err != nil {
return "", err
}
return stream + "_" + strings.ToUpper(kind) + "_worker", nil
}
// An ask id is a step id of the form buildstep.NewID gives: the time in milliseconds in base 36 (8 or 9 digits) and six
// random hex digits. Its form keeps it apart from a module's or a node's name in a key.
var askIDOK = regexp.MustCompile(`^[0-9a-z]{8,9}[0-9a-f]{6}$`)
func checkAskID(id string) error {
if !askIDOK.MatchString(id) {
return fmt.Errorf("delivery: %q is not an ask id (base-36 milliseconds and six hex digits)", id)
}
return nil
}
// StepAsk is the message on a build seat's stream: who asked what, of which kind and language, and the one whole job.
// Exactly one of Build and Check is set: a delivery ask is a build; a check ask a build (a candidate's fingerprint) or a
// check; a recheck ask a check; an image ask is always a build.
type StepAsk struct {
V int `json:"v"`
ID string `json:"id"`
Kind string `json:"kind"`
Language string `json:"language"`
// Asker is the module that put the ask (mesh-delivery-builder or mesh-delivery-checker); the stream's grant is what
// makes it true, this field only names it to a reader.
Asker string `json:"asker"`
At time.Time `json:"at"`
Build *BuildJob `json:"build,omitempty"`
Check *CheckJob `json:"check,omitempty"`
}
// BuildJob is a build ask's job: fetch the commit (unless every artifact is a copy), then build or copy each artifact,
// pushing it when Publish says so, and answer each artifact's digest and reference (BuildAnswer).
type BuildJob struct {
// Module is the module's name; Path its directory in the repository (the build tool's `module`, "." for its root).
Module string `json:"module"`
Path string `json:"path,omitempty"`
// FP is the fingerprint the build is of (the builder's key), named for a reader of the ask.
FP string `json:"fp,omitempty"`
// Fetch is the fetch tool's arguments, as the build module's fetch tool takes them (buildstep.FetchRequest's Args);
// absent when every artifact is a copy.
Fetch map[string]any `json:"fetch,omitempty"`
// Artifacts are built in this order.
Artifacts []AskArtifact `json:"artifacts"`
// Publish says the build pushes an image or publishes a package: a trunk build only (ADR 0487 clause 12).
Publish bool `json:"publish,omitempty"`
// Toolchain is the toolchain the fingerprint was computed with; a build module running another refuses the ask.
Toolchain *ToolchainUsed `json:"toolchain,omitempty"`
}
// AskArtifact is one artifact of a build ask: built from the fetched tree, or copied from From, an upstream pin (ADR
// 0498 clause 4).
type AskArtifact struct {
Artifact string `json:"artifact"`
From string `json:"from,omitempty"`
}
// CheckJob is a check ask's job, the levels of one check (ADR 0525 §3): fetch the head merged onto its trunk once, then
// for each level in order digest its packages, take the kept verdicts for the digests that have one, and test the rest,
// the packages the change's own tests are in first, stopping at the first level that does not pass (CheckAnswer).
type CheckJob struct {
// Fetch is the fetch tool's arguments (repository, commit, ref, base, and the build module's own: beside, package).
Fetch map[string]any `json:"fetch"`
// Levels are run in this order (LevelUnit, LevelIntegration); a try may name one.
Levels []string `json:"levels"`
// Kept says the build module takes the kept verdicts of the checker's kept bucket (CheckerKeptBucket, by KeptKey)
// for the packages whose digest has one and covers the siblings the fetch placed, and tests only the others (issue
// 1887's design limit 1). A try that runs some tests (Run), and a level the candidate's own build module judges
// (Runner), keep nothing.
Kept bool `json:"kept,omitempty"`
// Runner is the test step's runner ("workspace": the candidate's own build module judges it, ADR 0535 §4), used
// when the merge changes a path under RunnerWhen (every merge when RunnerWhen is empty).
Runner string `json:"runner,omitempty"`
RunnerWhen string `json:"runner_when,omitempty"`
// Run is a try's test pattern (go test -run): each level runs it alone, digesting nothing.
Run string `json:"run,omitempty"`
}
func (a StepAsk) check() error {
if err := checkAskID(a.ID); err != nil {
return err
}
if err := checkAskKind(a.Kind); err != nil {
return err
}
if err := checkAskLanguage(a.Language); err != nil {
return err
}
if err := checkName("asker", a.Asker); err != nil {
return err
}
if a.At.IsZero() {
return fmt.Errorf("delivery: the ask %s names no time", a.ID)
}
switch {
case (a.Build == nil) == (a.Check == nil):
return fmt.Errorf("delivery: the ask %s holds a build or a check, exactly one", a.ID)
case a.Build != nil && a.Kind == AskRecheck:
return fmt.Errorf("delivery: the ask %s is a build on the recheck stream, which carries checks alone", a.ID)
case a.Check != nil && a.Kind == AskDelivery:
return fmt.Errorf("delivery: the ask %s is a check on the delivery stream, which carries trunk builds alone", a.ID)
case a.Check != nil && a.Language == AskImage:
return fmt.Errorf("delivery: the ask %s is a check of an image, which has no tests", a.ID)
}
if a.Build != nil {
return a.Build.check(a.ID)
}
return a.Check.check(a.ID)
}
func (b BuildJob) check(id string) error {
if err := checkName("module", b.Module); err != nil {
return err
}
if len(b.Artifacts) == 0 {
return fmt.Errorf("delivery: the build ask %s names no artifact", id)
}
fetch := false
for _, a := range b.Artifacts {
if err := checkName("artifact", a.Artifact); err != nil {
return err
}
fetch = fetch || a.From == ""
}
if fetch {
if b.Path == "" {
return fmt.Errorf("delivery: the build ask %s names no directory (path) for the module %s", id, b.Module)
}
if err := checkFetchArgs(id, b.Fetch); err != nil {
return err
}
}
return nil
}
func (c CheckJob) check(id string) error {
if len(c.Levels) == 0 {
return fmt.Errorf("delivery: the check ask %s names no level", id)
}
for _, l := range c.Levels {
if err := checkLevel(l); err != nil {
return err
}
}
return checkFetchArgs(id, c.Fetch)
}
var askCommitOK = regexp.MustCompile(`^[0-9a-f]{40}$`)
// checkFetchArgs refuses fetch arguments with no repository or no full commit; the rest is the build module's fetch to
// judge, as a direct call's is.
func checkFetchArgs(id string, args map[string]any) error {
repository, _ := args["repository"].(string)
commit, _ := args["commit"].(string)
if repository == "" || !strings.Contains(repository, "/") {
return fmt.Errorf("delivery: the ask %s fetches no repository <owner>/<name>", id)
}
if !askCommitOK.MatchString(commit) {
return fmt.Errorf("delivery: the ask %s fetches no full commit", id)
}
return nil
}
// EncodeStepAsk is a as JSON at StepAskVersion; it refuses what DecodeStepAsk refuses.
func EncodeStepAsk(a StepAsk) ([]byte, error) {
a.V = StepAskVersion
a.At = a.At.UTC()
if err := a.check(); err != nil {
return nil, err
}
return json.Marshal(a)
}
// DecodeStepAsk reads an ask; it refuses an unknown version (ErrUnknownVersion), an unknown kind or language, an id that
// is not an ask id, a build and a check together or neither, a job on a stream that does not carry it, and a fetch with
// no repository or no full commit.
func DecodeStepAsk(data []byte) (StepAsk, error) {
var probe struct {
V *int `json:"v"`
}
if err := json.Unmarshal(data, &probe); err != nil {
return StepAsk{}, fmt.Errorf("delivery: an ask is not JSON: %w", err)
}
if probe.V == nil || *probe.V != StepAskVersion {
return StepAsk{}, fmt.Errorf("%w: an ask of version %v, this reader reads %d", ErrUnknownVersion, probe.V, StepAskVersion)
}
var a StepAsk
if err := json.Unmarshal(data, &a); err != nil {
return StepAsk{}, fmt.Errorf("delivery: an ask does not fit its shape: %w", err)
}
if err := a.check(); err != nil {
return StepAsk{}, err
}
return a, nil
}
// BuildAnswer is a build ask's step result: each artifact's digest and, for a pushed one, its reference.
type BuildAnswer struct {
Artifacts []AskBuilt `json:"artifacts"`
Toolchain *ToolchainUsed `json:"toolchain,omitempty"`
}
// AskBuilt is one artifact a build ask built.
type AskBuilt struct {
Artifact string `json:"artifact"`
Digest string `json:"digest"`
Reference string `json:"reference,omitempty"`
}
// CheckAnswer is a check ask's step result, each part the build module's own tool's answer whole: the fetch's, then one
// LevelAnswer per level run, in order; a level that did not pass is the last. The checker judges and keeps from it as it
// did from the tools' answers.
type CheckAnswer struct {
Fetch json.RawMessage `json:"fetch"`
Levels []LevelAnswer `json:"levels,omitempty"`
}
// LevelAnswer is one level of a check ask: the digest's answer (absent when nothing was digested: the level ran whole and
// keeps nothing), the kept verdicts taken, and one test result per test step run, in the order run.
type LevelAnswer struct {
Level string `json:"level"`
// Judged is what the digest under the candidate's runner said of the program that ran it ("candidate <merge>"),
// when the candidate judged the level (ADR 0535 §4, §5); empty otherwise.
Judged string `json:"judged,omitempty"`
Digest json.RawMessage `json:"digest,omitempty"`
Kept []KeptTaken `json:"kept,omitempty"`
Tests []json.RawMessage `json:"tests,omitempty"`
}
// KeptTaken is a package whose kept verdict the check took instead of testing it.
type KeptTaken struct {
Package string `json:"package"`
Digest string `json:"digest"`
Tests int `json:"tests"`
}
// AskStepKey is the key, in the build module's StepsBucket, of the step that runs an ask on node: `<node>.<ask id>`
// (ADR 0525 §6). The node is the one that pulled the ask, which the asker learns from the key.
func AskStepKey(node, askID string) (string, error) {
if err := checkAskID(askID); err != nil {
return "", err
}
return StepKey(node, askID)
}
// AskStepWatch is the pattern an asker watches its ask's step with, the node a wildcard: `*.<ask id>`.
func AskStepWatch(askID string) (string, error) {
if err := checkAskID(askID); err != nil {
return "", err
}
return "*." + askID, nil
}
// ParseAskStepKey reads the node and the ask id from a step key of an ask; it refuses a key that is not `<node>.<ask
// id>`, such as a have key `<node>.<module>` or a step key a direct tool call wrote under another id form.
func ParseAskStepKey(key string) (node, askID string, err error) {
node, askID, err = ParseStepKey(key)
if err != nil {
return "", "", err
}
if err := checkAskID(askID); err != nil {
return "", "", fmt.Errorf("delivery: %q is not an ask's step key: %w", key, err)
}
return node, askID, nil
}
// PauseKey is the key, in a build module's StepsBucket, that keeps whether the module on node takes asks:
// `paused.<node>` (issue 1887's design limit 2: every holder pulls from one shared worker, so a node's pause is its holder
// not fetching, kept here so a restart keeps it). A pause key is never an ask's step key: its second part is a node.
func PauseKey(node string) (string, error) {
if err := checkName("node", node); err != nil {
return "", err
}
return "paused." + node, nil
}
// Pause is the value of a pause key: whether the node's module takes no asks, who said so and when.
type Pause struct {
V int `json:"v"`
Paused bool `json:"paused"`
By string `json:"by,omitempty"`
At time.Time `json:"at"`
}
// EncodePause is p as JSON at Version.
func EncodePause(p Pause) ([]byte, error) {
p.V = Version
p.At = p.At.UTC()
return json.Marshal(p)
}
// DecodePause reads a pause value; it refuses an unknown version (ErrUnknownVersion).
func DecodePause(data []byte) (Pause, error) {
var p Pause
if err := decode(data, &p); err != nil {
return Pause{}, err
}
return p, nil
}
+38
View File
@@ -0,0 +1,38 @@
package delivery_test
import (
"testing"
"git.novox.be/novox/mesh-sdk/go/buildstep"
"git.novox.be/novox/mesh-sdk/go/delivery"
)
// 1887/A1b: the ask's step key is <node>.<ask id>, watched with *.<ask id>, and its parser refuses another bucket's key.
func TestAskStepKey(t *testing.T) {
id := buildstep.NewID() // an ask's id is a step id
key, err := delivery.AskStepKey("shanks", id)
if err != nil || key != "shanks."+id {
t.Fatalf("AskStepKey = %q, %v", key, err)
}
if w, err := delivery.AskStepWatch(id); err != nil || w != "*."+id {
t.Fatalf("AskStepWatch = %q, %v", w, err)
}
node, got, err := delivery.ParseAskStepKey(key)
if err != nil || node != "shanks" || got != id {
t.Fatalf("ParseAskStepKey = %q %q %v", node, got, err)
}
for _, other := range []string{
"shanks.mesh-delivery-view", // a have key
"shanks.node", // a node's own have key
"shanks.postgres", // a have key whose module is one token
"mesh-delivery-builder_builds.x", // a pulse key
"shanks",
} {
if _, _, err := delivery.ParseAskStepKey(other); err == nil {
t.Errorf("parsed %q", other)
}
}
if _, err := delivery.AskStepKey("shanks", "postgres"); err == nil {
t.Error("AskStepKey took a module's name for an ask id")
}
}
+148
View File
@@ -0,0 +1,148 @@
package delivery
import (
"errors"
"reflect"
"strings"
"testing"
"time"
)
const askCommit = "0123456789abcdef0123456789abcdef01234567"
func buildAsk() StepAsk {
return StepAsk{ID: "mh3k2x9qa1b2c3", Kind: AskDelivery, Language: AskGo, Asker: "mesh-delivery-builder",
At: time.Date(2026, 10, 23, 13, 0, 0, 0, time.UTC),
Build: &BuildJob{Module: "mesh-delivery-view", Path: "modules/mesh-delivery-view", FP: "ab12",
Fetch: map[string]any{"repository": "novox/mesh-delivery", "commit": askCommit, "dir": "modules/mesh-delivery-view", "paths": []any{"cmd"}},
Artifacts: []AskArtifact{{Artifact: "bundle"}, {Artifact: "base", From: "registry/x@sha256:00"}},
Publish: true, Toolchain: &ToolchainUsed{Version: "go1.26.0", Recipe: "3"}}}
}
func checkAsk() StepAsk {
return StepAsk{ID: "mh3k2x9qa1b2c4", Kind: AskCheck, Language: AskNode, Asker: "mesh-delivery-checker",
At: time.Date(2026, 10, 23, 13, 0, 0, 0, time.UTC),
Check: &CheckJob{Fetch: map[string]any{"repository": "novox/mesh-catalog", "commit": askCommit, "ref": "refs/pull/3/head", "base": askCommit},
Levels: []string{LevelUnit, LevelIntegration}, Kept: true, Runner: "workspace", RunnerWhen: "modules/mesh-build-go/", Run: ""}}
}
// 1887/A1a: an ask encodes and decodes with every field.
func TestAskRoundTrips(t *testing.T) {
for _, a := range []StepAsk{buildAsk(), checkAsk()} {
raw, err := EncodeStepAsk(a)
if err != nil {
t.Fatal(err)
}
got, err := DecodeStepAsk(raw)
if err != nil {
t.Fatal(err)
}
a.V = StepAskVersion
if !reflect.DeepEqual(got, a) {
t.Fatalf("round trip:\n got %+v\nwant %+v", got, a)
}
}
}
// 1887/A1a: an ask with an unknown kind or language is refused.
func TestAskRefusesUnknownKindAndLanguage(t *testing.T) {
cases := map[string]func(*StepAsk){
"kind": func(a *StepAsk) { a.Kind = "build" },
"language": func(a *StepAsk) { a.Language = "typescript" },
"id": func(a *StepAsk) { a.ID = "postgres" },
"neither": func(a *StepAsk) { a.Build = nil },
"both": func(a *StepAsk) { a.Check = &CheckJob{Levels: []string{LevelUnit}, Fetch: a.Build.Fetch} },
"recheck": func(a *StepAsk) { a.Kind = AskRecheck },
"commit": func(a *StepAsk) { a.Build.Fetch = map[string]any{"repository": "novox/x", "commit": "main"} },
"path": func(a *StepAsk) { a.Build.Path = "" },
}
for name, change := range cases {
a := buildAsk()
change(&a)
if _, err := EncodeStepAsk(a); err == nil {
t.Errorf("%s: encoded", name)
}
}
// A copy alone fetches nothing and needs no path.
a := buildAsk()
a.Build.Fetch, a.Build.Path, a.Build.Artifacts = nil, "", []AskArtifact{{Artifact: "base", From: "registry/x@sha256:00"}}
if _, err := EncodeStepAsk(a); err != nil {
t.Errorf("a copy alone: %v", err)
}
c := checkAsk()
c.Kind = AskDelivery
if _, err := EncodeStepAsk(c); err == nil {
t.Error("a check on the delivery stream encoded")
}
c = checkAsk()
c.Language = AskImage
if _, err := EncodeStepAsk(c); err == nil {
t.Error("a check of an image encoded")
}
raw, _ := EncodeStepAsk(buildAsk())
for _, bad := range []string{
strings.Replace(string(raw), `"kind":"delivery"`, `"kind":"queue"`, 1),
strings.Replace(string(raw), `"language":"go"`, `"language":"rust"`, 1),
} {
if _, err := DecodeStepAsk([]byte(bad)); err == nil {
t.Errorf("decoded %s", bad)
}
}
if _, err := DecodeStepAsk([]byte(strings.Replace(string(raw), `"v":1`, `"v":2`, 1))); !errors.Is(err, ErrUnknownVersion) {
t.Errorf("version 2: %v", err)
}
}
// 1887/A1a: the subjects, stream and workers are the seat's, as the broker derives them.
func TestAskSubjects(t *testing.T) {
s, err := AskSubject(AskGo, AskCheck)
if err != nil || s != "mesh.seat.node-build-go.accept.check" {
t.Fatalf("AskSubject = %q, %v", s, err)
}
language, kind, err := ParseAskSubject("mesh.seat.node-build-image.accept.delivery")
if err != nil || language != AskImage || kind != AskDelivery {
t.Fatalf("ParseAskSubject = %q %q %v", language, kind, err)
}
for _, bad := range []string{"mesh.seat.node-build-agent.accept.build", "mesh.seat.node-build-go.accept.queue", "mesh.seat.node-build-go.event.x"} {
if _, _, err := ParseAskSubject(bad); err == nil {
t.Errorf("parsed %s", bad)
}
}
if st, _ := AskStream(AskNode); st != "SEAT_NODE_BUILD_NODE" {
t.Errorf("AskStream = %q", st)
}
if w, _ := AskWorker(AskGo, AskRecheck); w != "SEAT_NODE_BUILD_GO_RECHECK_worker" {
t.Errorf("AskWorker = %q", w)
}
if seat, _ := BuildSeat(AskImage); seat != "node-build-image" {
t.Errorf("BuildSeat = %q", seat)
}
if m, _ := BuildModuleOf(AskNode); m != "mesh-build-node" {
t.Errorf("BuildModuleOf = %q", m)
}
if _, err := BuildSeat("agent"); err == nil {
t.Error("BuildSeat(agent)")
}
}
// 1887/A4b: a building key names the asks put for it; a done key or a name that is no ask id is refused.
func TestABuildingKeyNamesItsAsks(t *testing.T) {
b := BuilderBuild{State: BuildBuilding, By: "novox.builder", Asks: map[string]string{AskGo: "mh3k2x9qa1b2c3"}}
raw, err := EncodeBuilderBuild(b)
if err != nil {
t.Fatal(err)
}
got, err := DecodeBuilderBuild(raw)
if err != nil || len(got.Asks) != 1 || got.Asks[AskGo] != "mh3k2x9qa1b2c3" {
t.Fatalf("decoded %+v, %v", got, err)
}
for _, bad := range []BuilderBuild{
{State: BuildBuilding, Asks: map[string]string{AskGo: "postgres"}},
{State: BuildBuilding, Asks: map[string]string{"typescript": "mh3k2x9qa1b2c3"}},
{State: BuildFailed, Error: "x", Asks: map[string]string{AskGo: "mh3k2x9qa1b2c3"}},
} {
if _, err := EncodeBuilderBuild(bad); err == nil {
t.Errorf("encoded %+v", bad)
}
}
}
+55
View File
@@ -69,6 +69,61 @@ func SeatTake(worker string, handle func(Work) error) error {
return err
}
// Pulled is one message a bundle pulled from a seat's worker (SeatPull), named by the runtime's id for it, which
// SeatSettle gives its fate.
type Pulled struct {
ID string `json:"id"`
Worker string `json:"worker"`
Subject string `json:"subject"`
Body json.RawMessage `json:"body"`
Headers map[string]string `json:"headers,omitempty"`
Delivered uint64 `json:"delivered"`
}
// The fates SeatSettle gives a pulled message: done, handed back to be delivered again, dropped, or still being worked.
const (
SettleAck = "ack"
SettleNak = "nak"
SettleTerm = "term"
SettleProgress = "progress"
)
// SeatPull pulls at most one message from a held seat's worker the module's membership lists, waiting at most wait
// (none when 0, at most 25 s): nil when none came. Unlike SeatTake nothing is fetched ahead: a build module pulls an
// ask only when it has a free slot (novox/hq ADR 0525 §4; issue 1887 A11). A message never settled is delivered again
// by the worker's acknowledgement wait.
func SeatPull(worker string, wait time.Duration) (*Pulled, error) {
c, err := live()
if err != nil {
return nil, err
}
raw, err := c.ask("mesh/seat.pull", map[string]any{"worker": worker, "wait_ms": wait.Milliseconds()}, seatTimeout)
if err != nil {
return nil, err
}
if len(raw) == 0 || string(raw) == "null" {
return nil, nil
}
var p Pulled
if err := json.Unmarshal(raw, &p); err != nil {
return nil, err
}
if p.ID == "" {
return nil, nil
}
return &p, nil
}
// SeatSettle gives the message pulled under id its fate (SettleAck, SettleNak, SettleTerm, SettleProgress).
func SeatSettle(id, fate string) error {
c, err := live()
if err != nil {
return err
}
_, err = c.ask("mesh/seat.settle", map[string]any{"id": id, "fate": fate}, seatTimeout)
return err
}
// SeatRecord reads one key of a record a seat keeps for this module, under its own name, or nil.
func SeatRecord(bucket, key string) (*StateEntry, error) {
c, err := live()
+65
View File
@@ -0,0 +1,65 @@
package stdio
import (
"bufio"
"encoding/json"
"io"
"testing"
"time"
)
// 1887/A11: a bundle pulls one message through its runtime, naming the worker and the wait, and settles it by the id the
// runtime gave.
func TestABundlePullsAndSettlesThroughTheRuntime(t *testing.T) {
t.Setenv(ServedAs, "mesh-build-go")
inR, inW := io.Pipe()
outR, outW := io.Pipe()
got := make(chan *Pulled, 1)
go func() {
_ = ServeOn("", []Tool{{Name: "start", Run: func(map[string]any) (any, error) {
p, err := SeatPull("SEAT_NODE_BUILD_GO_DELIVERY_worker", 1500*time.Millisecond)
if err != nil {
return nil, err
}
got <- p
return map[string]any{}, SeatSettle(p.ID, SettleAck)
}}}, inR, outW)
}()
lines := bufio.NewScanner(outR)
next := func() map[string]any {
done := make(chan map[string]any, 1)
go func() {
if lines.Scan() {
var m map[string]any
_ = json.Unmarshal(lines.Bytes(), &m)
done <- m
}
}()
select {
case m := <-done:
return m
case <-time.After(3 * time.Second):
t.Fatal("the bundle said nothing")
return nil
}
}
send := func(m map[string]any) { b, _ := json.Marshal(m); _, _ = inW.Write(append(b, '\n')) }
send(map[string]any{"jsonrpc": "2.0", "id": 1, "method": "tools/call", "params": map[string]any{"name": "start"}})
asked := next()
params, _ := asked["params"].(map[string]any)
if asked["method"] != "mesh/seat.pull" || params["worker"] != "SEAT_NODE_BUILD_GO_DELIVERY_worker" || params["wait_ms"] != 1500.0 {
t.Fatalf("asked %v", asked)
}
send(map[string]any{"jsonrpc": "2.0", "id": asked["id"], "result": map[string]any{"id": "p1", "worker": "SEAT_NODE_BUILD_GO_DELIVERY_worker",
"subject": "mesh.seat.node-build-go.accept.delivery", "body": map[string]any{"v": 1}, "delivered": 1}})
p := <-got
if p.ID != "p1" || string(p.Body) != `{"v":1}` {
t.Fatalf("pulled %+v", p)
}
asked = next()
params, _ = asked["params"].(map[string]any)
if asked["method"] != "mesh/seat.settle" || params["id"] != "p1" || params["fate"] != "ack" {
t.Fatalf("settled %v", asked)
}
send(map[string]any{"jsonrpc": "2.0", "id": asked["id"], "result": map[string]any{}})
}