Compare commits

...
6 Commits
Author SHA1 Message Date
jschoubben c8bcb2cdf5 mesh-build-go asks a run's input of the seat verb the ask names, and reads no checker state (issue 1921, B0 review round 1)
mesh/delivery ready: it is merged on the Approve
mesh/check checking c8bcb2cdf567 on 980ade933853 | novox/mesh-catalog#622 in group feat/b0-checking-program
mesh/delivery-group group feat/b0-checking-program checking: 3 of 3 member(s) ready
The operator's answer: reading mesh-delivery-checker.inputs broke ADR 0596 §2. The manifest drops that read and invokes
seat:mesh-delivery-checker.input; the run step asks the verb through the runtime (stdio.Ask) and takes only the bytes
the ask names (delivery.RunInput.Read).

Claude-Session: https://claude.ai/code/session_01NcbgfFSnn24M3GcqttoU7i
2026-11-06 23:43:05 +01:00
jschoubben 608c4d4be6 Fail a run step whose answer is over the bus's value limit, naming its size (issue 1921, review round 1)
Claude-Session: https://claude.ai/code/session_01NcbgfFSnn24M3GcqttoU7i
2026-11-06 23:26:12 +01:00
jschoubben da1356cd7c Bind a run's heads read-only over its workspace, read its input from the state the ask names, and prove the kept programs' sweep (issue 1921, review round 1)
Claude-Session: https://claude.ai/code/session_01NcbgfFSnn24M3GcqttoU7i
2026-11-06 23:24:48 +01:00
jschoubben fa73674e22 A run ask builds a checking program once and runs it over its heads and input; compose keeps seats (issue 1921, C1-C3)
Claude-Session: https://claude.ai/code/session_01NcbgfFSnn24M3GcqttoU7i
2026-11-06 22:20:25 +01:00
jschoubben 428ef8bbe5 Kept composers and programs get a list tool, a drop tool and a sweep (issue 1921, seams must 2)
Claude-Session: https://claude.ai/code/session_01NcbgfFSnn24M3GcqttoU7i
2026-11-06 22:05:05 +01:00
jschoubben 0220dfa92b Answer an ask the build module cannot read on its step key as refused (issue 1921, seams must 1)
Claude-Session: https://claude.ai/code/session_01NcbgfFSnn24M3GcqttoU7i
2026-11-06 22:03:51 +01:00
18 changed files with 763 additions and 29 deletions
+4
View File
@@ -49,6 +49,10 @@ already gone is no error, one a running step holds is refused); its caller calls
failed fetch removes its own workspace. A workspace no caller released is removed two hours after its last use (ADR
0509), and the removal is said in the log as the caller's fault.
The programs this module builds from a commit and keeps in its cache (composers of compose asks, checking programs of run
asks) are read by `build_go_kept` (kind, key, size, last use) and removed by `build_go_kept_drop {kind, key}`; one no ask used
for 30 days is swept with the workspaces (issue 1921).
A pin on a pull request's head (issue 1148): when `go mod download` fails on a `git.novox.be/novox/*` pseudo-version, the
fetch looks for that commit among the forge's `refs/pull/*/head`, places the repository at it under the workspace's
`pinned/beside/` as a plain directory, and hands it to Go (the fetch's commands, and the build with `GOPROXY=off`) through a
+14 -8
View File
@@ -27,16 +27,19 @@ const Bubblewrap = "bwrap"
// - the system's programs and libraries read-only, and of /etc only what a program needs to run (users, libraries,
// certificates, time zone), with a hosts file naming only the loopback;
// - the Workspace and Writable directories read-write, the ReadOnly ones (the toolchain, the modules the fetch step
// downloaded) read-only;
// downloaded) read-only, and the ReadOnlyInside ones (inside the Workspace) read-only over it;
// - the shared warm Caches read-only under a throwaway writable layer (S18), or read-write for a trunk build alone
// (WriteCaches), so a candidate's code never changes what a later build reads.
type Isolation struct {
Workspace string
Writable []string
ReadOnly []string
Caches []string
WriteCaches bool
Env []string
Workspace string
Writable []string
ReadOnly []string
// ReadOnlyInside are directories inside the Workspace or a Writable one bound read-only after them, so they stay
// read-only (a later bind of a path wins; a ReadOnly one inside the Workspace would be covered by it).
ReadOnlyInside []string
Caches []string
WriteCaches bool
Env []string
// Hosts is the file bound as /etc/hosts; Isolated writes one naming only the loopback.
Hosts string
}
@@ -91,7 +94,7 @@ func (iso Isolation) command(env []string, name string, args ...string) ([]strin
for _, group := range []struct {
what string
dirs []string
}{{"writable directory", iso.Writable}, {"read-only directory", iso.ReadOnly}, {"cache", iso.Caches}} {
}{{"writable directory", iso.Writable}, {"read-only directory", iso.ReadOnly}, {"read-only directory", iso.ReadOnlyInside}, {"cache", iso.Caches}} {
for _, d := range group.dirs {
if err := checkDir(group.what, d); err != nil {
return nil, err
@@ -118,6 +121,9 @@ func (iso Isolation) command(env []string, name string, args ...string) ([]strin
for _, d := range iso.Writable {
argv = append(argv, "--bind", d, d)
}
for _, d := range iso.ReadOnlyInside {
argv = append(argv, "--ro-bind", d, d)
}
if iso.Hosts != "" {
argv = append(argv, "--ro-bind", iso.Hosts, "/etc/hosts")
}
+34 -3
View File
@@ -561,6 +561,36 @@ func (p *Puller) declineRun(msg *Pulled, ask delivery.StepAsk, key string, step
p.mu.Unlock()
}
// refuseUnread ends an ask this module cannot read: when its data holds an id that keys a step, the step is written
// refused with the reason, so the asker learns at once that no node began it; otherwise (no id, not JSON) the ask is only
// dropped and logged. The message is dropped either way.
func (p *Puller) refuseUnread(msg *Pulled, why error) {
var probe struct {
ID string `json:"id"`
Kind string `json:"kind"`
}
if json.Unmarshal(msg.Data, &probe) == nil && probe.ID != "" {
if key, kerr := delivery.AskStepKey(p.Node, probe.ID); kerr == nil {
tool := "ask"
if probe.Kind != "" {
tool = probe.Kind + " ask"
}
at := now().UTC()
step := delivery.Step{ID: probe.ID, Tool: tool, Node: p.Node, State: delivery.StepRefused, Started: at, Ended: at,
Reason: fmt.Sprintf("this build module on %s cannot read the ask: %v", p.Node, why), Owner: sdkstep.Owner}
if werr := writeEnd(p.Put, key, step); werr != nil {
p.logf("the unread ask %s cannot be answered on %s: %v", probe.ID, p.Node, werr)
} else {
p.logf("an ask refused: %v", why)
_ = msg.Term()
return
}
}
}
p.logf("an ask dropped: %v", why)
_ = msg.Term()
}
func withKind(msg *Pulled, kind string) *Pulled {
if msg != nil {
msg.Kind = kind
@@ -576,9 +606,10 @@ func (p *Puller) take(ctx context.Context, msg *Pulled) {
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()
// An ask that is not one runs nowhere: delivering it again would only fail again. But an ask that names an id
// is answered on its step key as refused, naming why (a job this build module does not know, a version it does
// not read): a dropped ask must always be answered, or its asker waits to its bound for nothing.
p.refuseUnread(msg, err)
return
}
if lacks := delivery.Lacks(ask.Needs, p.Has); len(lacks) > 0 {
@@ -519,3 +519,39 @@ func TestAPoisonedAskIsSaidAndRunNoMore(t *testing.T) {
t.Errorf("acked %v naked %v, want it termed alone", acked, naked)
}
}
// An ask whose job this build module does not know is answered on its step key as refused, naming why, never only dropped
// (a dropped ask leaves its asker waiting to its bound); one with no id is still dropped.
func TestPullAnswersAnAskItCannotRead(t *testing.T) {
src, st := NewFakeAsks(), newSteps()
good := testAsk(delivery.AskCheck)
raw, err := delivery.EncodeStepAsk(good)
if err != nil {
t.Fatal(err)
}
var m map[string]any
_ = json.Unmarshal(raw, &m)
delete(m, "check")
m["frobnicate"] = map[string]any{"program": "x"}
odd, _ := json.Marshal(m)
// The job is not one this reader holds, so the ask carries none it knows: decode refuses it.
if _, err := delivery.DecodeStepAsk(odd); err == nil {
t.Fatal("the fixture ask decodes")
}
src.Publish(delivery.AskCheck, odd)
src.Publish(delivery.AskCheck, []byte(`{"v":1}`))
p := &Puller{Source: src, Language: delivery.AskGo, Node: "ace", Put: st.put, Capacity: func() int { return 1 },
PullWait: 10 * time.Millisecond, Run: func(ctx context.Context, a delivery.StepAsk) (any, error) {
t.Error("an unread ask ran")
return nil, nil
}}
ctx, cancel := context.WithCancel(context.Background())
defer cancel()
go p.Pull(ctx)
waitFor(t, "both termed", func() bool { _, tm, _ := src.Fates(); return len(tm) == 2 })
key, _ := delivery.AskStepKey("ace", good.ID)
s, ok := st.get(key)
if !ok || s.State != delivery.StepRefused || !strings.Contains(s.Reason, "cannot read the ask") || !strings.Contains(s.Reason, "exactly one") {
t.Fatalf("the unread ask's step = %+v (%v)", s, ok)
}
}
@@ -121,6 +121,9 @@ func (m *Module) Puller(steps []StepTool, source buildstep.AskSource) *buildstep
if a.Compose != nil {
return m.compose(ctx, ask.Fetch, a)
}
if a.Run != nil {
return m.run(ctx, ask.Fetch, a)
}
return buildstep.RunAsk(ctx, ask, a)
}}
}
@@ -38,7 +38,7 @@ var composerLocks sync.Map
// composerPath is where the composer of key is kept: the cache directory, beside Go's own caches.
func (m *Module) composerPath(key string) string {
return filepath.Join(m.Cache, "composers", key, "mesh-compose")
return m.programPath("composers", key)
}
// compose runs a compose ask: the composer of its commit, built once, over its request with the manifests filled.
@@ -98,17 +98,25 @@ func (m *Module) manifestOf(digest string) ([]byte, error) {
// composerOf is the composer kept under key, or, when none is, the one built now from the fetched commit (built true).
func (m *Module) composerOf(ctx context.Context, fetch buildstep.StepFunc, job delivery.ComposeJob, key string) (string, bool, error) {
exe := m.composerPath(key)
mu, _ := composerLocks.LoadOrStore(exe, &sync.Mutex{})
mu.(*sync.Mutex).Lock()
defer mu.(*sync.Mutex).Unlock()
return m.programOf(ctx, fetch, "composers", "composer", key, job.Fetch, job.Dir, job.Package)
}
// programOf is the program kept under key in the cache directory kind, or, when none is, the one built now: the commit
// fetch names is fetched (the fetch step: the forge key and the network, none of the change's code), and package in the
// Go module of dir is built in a run step shut off with nothing fetched. what names it in a failure.
func (m *Module) programOf(ctx context.Context, fetch buildstep.StepFunc, kind, what, key string, fetchArgs map[string]any, dir, pkg string) (string, bool, error) {
exe := m.programPath(kind, key)
mu := lockOf(exe)
mu.Lock()
defer mu.Unlock()
if st, err := os.Stat(exe); err == nil && st.Mode().IsRegular() {
touch(exe, m.clock())
return exe, false, nil
}
if fetch == nil {
return "", false, buildstep.Refuse("this build module fetches nothing")
}
out, err := fetch(ctx, job.Fetch)
out, err := fetch(ctx, fetchArgs)
if err != nil {
return "", false, err
}
@@ -125,11 +133,11 @@ func (m *Module) composerOf(ctx context.Context, fetch buildstep.StepFunc, job d
defer m.releaseSaid(f.Workspace)
done := m.spaces().Hold(f.Workspace)
defer done()
dir := filepath.Join(m.Work, f.Workspace, "tree", job.Dir)
if _, err := os.Stat(filepath.Join(dir, "go.mod")); err != nil {
return "", false, fmt.Errorf("%s at %s holds no go.mod in %s, so no composer can be built there", job.Fetch["repository"], job.Fetch["commit"], job.Dir)
tree := filepath.Join(m.Work, f.Workspace, "tree", dir)
if _, err := os.Stat(filepath.Join(tree, "go.mod")); err != nil {
return "", false, fmt.Errorf("%s at %s holds no go.mod in %s, so no %s can be built there", fetchArgs["repository"], fetchArgs["commit"], dir, what)
}
into := filepath.Join(m.Work, f.Workspace, "composer", "mesh-compose")
into := filepath.Join(m.Work, f.Workspace, what, "built")
if err := os.MkdirAll(filepath.Dir(into), 0o700); err != nil {
return "", false, err
}
@@ -137,15 +145,15 @@ func (m *Module) composerOf(ctx context.Context, fetch buildstep.StepFunc, job d
if err != nil {
return "", false, err
}
env := append(m.goEnvFor(dir), "GOPROXY=off", "GOTOOLCHAIN=local", "CGO_ENABLED=0")
r := run(ctx, env, "go", "-C", dir, "build", "-trimpath", "-o", into, job.Package)
env := append(m.goEnvFor(tree), "GOPROXY=off", "GOTOOLCHAIN=local", "CGO_ENABLED=0")
r := run(ctx, env, "go", "-C", tree, "build", "-trimpath", "-o", into, pkg)
if r.Status != 0 || r.Err != "" {
// The words Go says when the module holds no such package (a mesh-sdk from before the composer) are kept whole,
// so the asker tells that apart from a fault.
return "", false, fmt.Errorf("the composer %s does not build at %s: %s", job.Package, job.Fetch["commit"], compileError(r.Err+"\n"+r.Stderr+"\n"+r.Stdout))
return "", false, fmt.Errorf("the %s %s does not build at %s: %s", what, pkg, fetchArgs["commit"], compileError(r.Err+"\n"+r.Stderr+"\n"+r.Stdout))
}
if err := copyExecutable(into, exe); err != nil {
return "", false, fmt.Errorf("the composer could not be kept: %w", err)
return "", false, fmt.Errorf("the %s could not be kept: %w", what, err)
}
return exe, true, nil
}
@@ -150,11 +150,17 @@ func TestAComposeAskFailsLoudly(t *testing.T) {
}
}
// 1622: the module reads the delivery indexer's manifests, which a compose ask names by digest.
// 1622: the module reads the delivery indexer's manifests, which a compose ask names by digest. 1921 (B0): a run's input
// is asked of the checker seat's verb input, as the manifest invokes it, and no state of the checker's inputs is read
// (ADR 0596 §2).
func TestTheModuleReadsTheIndexersManifests(t *testing.T) {
if m := readManifest(t); !slices.Contains(m.Reads, ManifestsState) {
m := readManifest(t)
if !slices.Contains(m.Reads, ManifestsState) {
t.Errorf("reads = %v, lacks %s", m.Reads, ManifestsState)
}
if !slices.Contains(m.Invokes, runInputVerb) || slices.Contains(m.Reads, "mesh-delivery-checker.inputs") {
t.Errorf("invokes = %v, reads = %v: want %s invoked and no inputs state read", m.Invokes, m.Reads, runInputVerb)
}
}
// ADR 0593 §7 (C9): a compose ask naming the node-engine's repository builds its composer command in the Go module at the
@@ -348,6 +348,7 @@ func (m *Module) goEnvFor(dir string) []string {
// net for a caller that never released one (issue 1839, ADR 0509).
func (m *Module) CleanWorkspaces(now time.Time) {
m.sweepSaid(m.spaces().Sweep(now, buildstep.UnheldBound))
m.SweepKept(now, KeptBound)
}
// clock is the time now: Now when a test set it, else the wall clock.
@@ -0,0 +1,130 @@
package main
// **Kept programs have tools and a sweep** (novox/hq issue 1921, the seams review's must 2): the programs this module builds
// from a commit and keeps under its cache so each commit is built once on a node, the composers of compose asks and the
// programs of run asks, are read by build_go_kept, removed one by one by build_go_kept_drop, and swept when unused for
// KeptBound. A use touches the program's file, so its modification time is its last use.
import (
"fmt"
"os"
"path/filepath"
"sort"
"sync"
"time"
stdio "git.novox.be/novox/mesh-sdk/go"
)
// The tools of the kept programs.
const (
KeptTool = "build_go_kept"
KeptDropTool = "build_go_kept_drop"
)
// KeptBound is how long a kept program that no ask used is kept: a month, longer than any group lives, so only the build
// of a commit nobody asks about any more goes.
const KeptBound = 30 * 24 * time.Hour
// keptKinds are the cache directories of kept programs and the name of the program in each key's directory.
var keptKinds = map[string]string{"composers": "mesh-compose", "programs": "program"}
// KeptProgram is one kept program: its kind (the cache directory), its key, its size and when an ask last used it.
type KeptProgram struct {
Kind string `json:"kind"`
Key string `json:"key"`
Bytes int64 `json:"bytes"`
LastUsed time.Time `json:"last-used"`
}
// programPath is where the program of kind and key is kept.
func (m *Module) programPath(kind, key string) string {
return filepath.Join(m.Cache, kind, key, keptKinds[kind])
}
// lockOf is the mutex that guards the program at path against a build or a use while it is dropped.
func lockOf(path string) *sync.Mutex {
l, _ := composerLocks.LoadOrStore(path, &sync.Mutex{})
return l.(*sync.Mutex)
}
// touch records a use of the kept program at path.
func touch(path string, at time.Time) { _ = os.Chtimes(path, at, at) }
// Kept lists the kept programs, composers then programs, each kind by key.
func (m *Module) Kept() []KeptProgram {
out := []KeptProgram{}
for _, kind := range []string{"composers", "programs"} {
entries, _ := os.ReadDir(filepath.Join(m.Cache, kind))
for _, e := range entries {
if !e.IsDir() {
continue
}
st, err := os.Stat(m.programPath(kind, e.Name()))
if err != nil || !st.Mode().IsRegular() {
continue
}
out = append(out, KeptProgram{Kind: kind, Key: e.Name(), Bytes: st.Size(), LastUsed: st.ModTime().UTC()})
}
}
return out
}
// DropKept removes the kept program of kind and key; one already gone is no error.
func (m *Module) DropKept(kind, key string) error {
if _, ok := keptKinds[kind]; !ok {
return fmt.Errorf("a kept program is of kind composers or programs, not %q", kind)
}
if key == "" || key != filepath.Base(key) || key == "." || key == ".." {
return fmt.Errorf("%q is no kept program's key", key)
}
exe := m.programPath(kind, key)
l := lockOf(exe)
l.Lock()
defer l.Unlock()
return os.RemoveAll(filepath.Dir(exe))
}
// SweepKept removes the kept programs no ask used within bound before now, and answers what it removed.
func (m *Module) SweepKept(now time.Time, bound time.Duration) []KeptProgram {
var gone []KeptProgram
for _, k := range m.Kept() {
if now.Sub(k.LastUsed) <= bound {
continue
}
if m.DropKept(k.Kind, k.Key) == nil {
gone = append(gone, k)
}
}
sort.Slice(gone, func(i, j int) bool { return gone[i].Key < gone[j].Key })
if len(gone) > 0 && m.Say != nil {
m.Say("removed %d kept programs no ask used within %s", len(gone), bound)
}
return gone
}
// keptTools are the tools that read and drop the kept programs: no guard and no floor, since they only free room.
func (m *Module) keptTools() []stdio.Tool {
return []stdio.Tool{
{Name: KeptTool,
Description: "The programs this node keeps, built from a commit so each commit is built once: the composers of compose asks and " +
"the checking programs of run asks, each with its kind, key, size in bytes and when an ask last used it. One no ask used " +
"for 30 days is removed. Read-only.",
Run: func(map[string]any) (any, error) { return map[string]any{"kept": m.Kept()}, nil }},
{Name: KeptDropTool,
Description: "Remove one kept program by its kind (composers or programs) and key, as build_go_kept names them: the next ask " +
"for that commit builds it again. One already gone is no error.",
Input: map[string]any{"type": "object", "properties": map[string]any{
"kind": map[string]any{"type": "string", "enum": []string{"composers", "programs"}},
"key": map[string]any{"type": "string", "description": "the key build_go_kept names"},
}, "required": []string{"kind", "key"}},
Run: func(args map[string]any) (any, error) {
kind, _ := args["kind"].(string)
key, _ := args["key"].(string)
if err := m.DropKept(kind, key); err != nil {
return nil, err
}
return map[string]any{"dropped": key, "kind": kind}, nil
}},
}
}
@@ -0,0 +1,70 @@
package main
import (
"os"
"path/filepath"
"testing"
"time"
)
func keepProgram(t *testing.T, m *Module, kind, key string, size int, used time.Time) {
t.Helper()
exe := m.programPath(kind, key)
if err := os.MkdirAll(filepath.Dir(exe), 0o700); err != nil {
t.Fatal(err)
}
if err := os.WriteFile(exe, make([]byte, size), 0o700); err != nil {
t.Fatal(err)
}
touch(exe, used)
}
func toolOf(t *testing.T, m *Module, name string) func(map[string]any) (any, error) {
t.Helper()
for _, tool := range m.Tools(m.StepTools()) {
if tool.Name == name {
return tool.Run
}
}
t.Fatalf("no tool %s", name)
return nil
}
// Seams review must 2 (issue 1921): the kept composers and programs are listed with their size and last use, dropped one
// by one, and swept with the workspaces when unused past KeptBound; a use keeps a program.
func TestKeptProgramsAreListedDroppedAndSwept(t *testing.T) {
m, _ := fetchModule(t, &fakeStep{})
now := time.Now().UTC().Truncate(time.Second)
keepProgram(t, m, "composers", "old", 10, now.Add(-KeptBound-time.Hour))
keepProgram(t, m, "programs", "recent", 20, now.Add(-time.Hour))
keepProgram(t, m, "programs", "gone", 30, now)
out, err := toolOf(t, m, KeptTool)(nil)
if err != nil {
t.Fatal(err)
}
list := out.(map[string]any)["kept"].([]KeptProgram)
if len(list) != 3 || list[0].Kind != "composers" || list[0].Key != "old" || list[0].Bytes != 10 || !list[0].LastUsed.Equal(now.Add(-KeptBound-time.Hour)) {
t.Fatalf("the kept programs = %+v", list)
}
for _, bad := range []map[string]any{{"kind": "nope", "key": "gone"}, {"kind": "programs", "key": "../composers"}, {"kind": "programs", "key": ""}} {
if _, err := toolOf(t, m, KeptDropTool)(bad); err == nil {
t.Errorf("dropped with %v", bad)
}
}
for i := 0; i < 2; i++ { // a second drop is no error
if _, err := toolOf(t, m, KeptDropTool)(map[string]any{"kind": "programs", "key": "gone"}); err != nil {
t.Fatalf("drop %d: %v", i, err)
}
}
if _, err := os.Stat(m.programPath("programs", "gone")); !os.IsNotExist(err) {
t.Error("the dropped program is still kept")
}
// The sweep runs with the workspaces' (CleanWorkspaces, on its timer).
m.CleanWorkspaces(now)
if got := m.Kept(); len(got) != 1 || got[0].Key != "recent" {
t.Fatalf("kept after the sweep = %+v", got)
}
}
@@ -58,6 +58,7 @@ func main() {
m.Settings = m.Home + "/settings.json"
m.ReadKept = readOf(stdio.State(KeptState)).Get
m.ReadManifest = readOf(stdio.State(ManifestsState)).Get
m.AskInput = stdio.Ask
m.Probe = ProbeGo(m.Home, m.Cache)
m.Bubblewrap = ProbeBubblewrap
// Outside a launch by the runtime — a build, a check — there is no node and no state to write.
@@ -93,7 +93,12 @@ type Module struct {
// ReadManifest reads a key of the delivery indexer's manifests state (ManifestsState), which a compose ask's request
// names its manifests by (compose.go); GoModCache, when set, is the Go module cache in place of Cache's (a test).
ReadManifest func(key string) (json.RawMessage, error)
GoModCache string
// AskInput asks a seat verb the module invokes (module.json) as the runtime calls it (stdio.Ask): the verb a run ask's
// input names, answering the input by its digest (delivery.RunInput.Read).
AskInput func(key string, body any) (json.RawMessage, error)
// PlaceHead, when set, lays a run's head at a directory instead of fetching it (a test).
PlaceHead func(ctx context.Context, h delivery.RunHead, dir string) error
GoModCache string
spacesOnce sync.Once
workspaces *buildstep.Workspaces
@@ -115,7 +115,7 @@ func TestAGoThatCannotBeRunRefusesEveryStepAndHealthIsUnhealthy(t *testing.T) {
{Name: "short_probe", Run: run},
}, m.StepTools()...)
for _, tool := range m.Tools(steps) {
if tool.Name == StatusTool || tool.Name == HealthTool || tool.Name == StepReadTool || tool.Name == StepsReadTool || tool.Name == ToolchainsReadTool || tool.Name == ReleaseTool || tool.Name == StepEndTool {
if tool.Name == StatusTool || tool.Name == HealthTool || tool.Name == StepReadTool || tool.Name == StepsReadTool || tool.Name == ToolchainsReadTool || tool.Name == ReleaseTool || tool.Name == KeptTool || tool.Name == KeptDropTool || tool.Name == StepEndTool {
continue
}
var reason string
@@ -0,0 +1,132 @@
package main
// **A run ask runs a program built from a change, over heads it reads** (novox/hq ADR 0644, ADR 0656; issue 1921): the
// checking program of a group's check. The module builds the program at the commit the ask names, once per
// delivery.RunKey, as a compose ask builds a composer (programOf); fetches every head the ask names and lays each read-only
// under the directory MESH_RUN_HEADS names, at `<owner>/<name>`; asks the seat verb the ask's input names for it by its
// digest (as module.json invokes it; never a read of the asker's state, ADR 0596 §2) and writes it to the file
// MESH_RUN_INPUT names; and runs the program shut off with the ask's
// arguments: no network, none of the account. The answer is delivery.RunAnswer: the program's exit, the last line of its
// standard output and the first line of its standard error. A program that fails is an answer, not a fault of the step.
import (
"context"
"encoding/json"
"errors"
"fmt"
"os"
"path/filepath"
"strings"
"git.novox.be/novox/mesh-catalog/modules/mesh-build-go/buildstep"
sdkstep "git.novox.be/novox/mesh-sdk/go/buildstep"
"git.novox.be/novox/mesh-sdk/go/delivery"
)
// run runs a run ask: the program of its commit, built once, over its heads and its input.
func (m *Module) run(ctx context.Context, fetch buildstep.StepFunc, a delivery.StepAsk) (any, error) {
job := *a.Run
body, err := m.inputOf(job.Input)
if err != nil {
return nil, err
}
installed, err := m.Probe(ctx)
if err != nil {
return nil, err
}
key := delivery.RunKey(job.Program, delivery.ToolchainUsed{Version: installed, Recipe: m.Recipe.Version()}, a.Packages)
fetchArgs := sdkstep.FetchRequest{Repository: job.Program.Repository, Commit: job.Program.Commit, Packages: a.Packages}.Args()
exe, built, err := m.programOf(ctx, fetch, "programs", "checking program", key, fetchArgs, job.Program.Dir, job.Program.Package)
if err != nil {
return nil, err
}
ans, err := m.runProgram(ctx, exe, job, body)
if err != nil {
return nil, err
}
ans.Built, ans.Key = built, key
// The answer is written on the step key as one value: one over the bus's value limit (the input's bound) fails the
// step, naming its size, before it is written.
if raw, err := json.Marshal(ans); err != nil {
return nil, err
} else if len(raw) > delivery.RunInputBound {
return nil, fmt.Errorf("the checking program's answer is %d bytes, over the bound of %d bytes a step key holds", len(raw), delivery.RunInputBound)
}
return ans, nil
}
// inputOf is the input a run names, asked of the seat verb the ask names (module.json invokes it) and checked against
// its digest and size: bytes that are not the input named refuse the ask.
func (m *Module) inputOf(in delivery.RunInput) ([]byte, error) {
if m.AskInput == nil {
return nil, errors.New("this module asks no seat verb for a run's input")
}
raw, err := in.Read(m.AskInput)
if err != nil && raw != nil {
return nil, buildstep.Refuse("%v", err)
}
return raw, err
}
// placeHead lays one head's commit as a plain directory at dir, with no symlink in it; a test sets its own.
func (m *Module) placeHead(ctx context.Context, h delivery.RunHead, dir string) error {
if m.PlaceHead != nil {
return m.PlaceHead(ctx, h, dir)
}
if err := os.MkdirAll(dir, 0o700); err != nil {
return err
}
if err := buildstep.FetchCommit(ctx, m.Run, dir, m.remote(h.Repository), h.Commit, "", m.fetchOptions(ctx, h.Repository, h.Commit, "")...); err != nil {
return err
}
if err := m.dissociate(ctx, dir); err != nil {
return err
}
if link := firstSymlink(dir); link != "" {
return fmt.Errorf("%s holds a symlink, %s", h.Repository, link)
}
return nil
}
// runProgram lays the heads and the input in a workspace of its own and runs exe shut off over them.
func (m *Module) runProgram(ctx context.Context, exe string, job delivery.RunJob, input []byte) (delivery.RunAnswer, error) {
id := sdkstep.NewID()
done := m.spaces().Hold(id)
defer m.releaseSaid(id)
defer done()
ws := filepath.Join(m.Work, id)
heads := filepath.Join(ws, "heads")
for _, h := range job.Heads {
if err := m.placeHead(ctx, h, filepath.Join(heads, filepath.FromSlash(h.Repository))); err != nil {
return delivery.RunAnswer{}, fmt.Errorf("the head of %s at %s: %w", h.Repository, h.Commit, err)
}
}
if err := os.MkdirAll(heads, 0o700); err != nil {
return delivery.RunAnswer{}, err
}
file := filepath.Join(ws, "input")
if err := os.WriteFile(file, input, 0o600); err != nil {
return delivery.RunAnswer{}, err
}
run := m.Run
if !m.Inside {
iso := m.isolation(id, false)
iso.ReadOnly = append(iso.ReadOnly, exe)
iso.ReadOnlyInside = []string{heads} // bound after the workspace, so the program cannot change a head
run = buildstep.Isolated(m.Run, iso)
}
r := run(ctx, []string{delivery.RunHeadsEnv + "=" + heads, delivery.RunInputEnv + "=" + file}, exe, job.Args...)
if r.Err != "" && r.Status == 0 {
return delivery.RunAnswer{}, fmt.Errorf("the checking program could not run: %s", r.Err)
}
ans := delivery.RunAnswer{Exit: r.Status, Said: firstLine(r.Err + "\n" + r.Stderr)}
line := strings.TrimSpace(lastLine(r.Stdout))
switch {
case line == "":
case json.Valid([]byte(line)):
ans.Output = json.RawMessage(line)
default:
ans.Output, _ = json.Marshal(line)
}
return ans, nil
}
@@ -0,0 +1,141 @@
//go:build integration
package main
import (
"context"
"encoding/json"
"errors"
"net"
"os"
"path/filepath"
"strings"
"testing"
"time"
"github.com/nats-io/nats.go"
"github.com/nats-io/nats.go/jetstream"
"git.novox.be/novox/mesh-catalog/modules/mesh-build-go/buildstep"
"git.novox.be/novox/mesh-sdk/go/delivery"
"git.novox.be/novox/mesh-sdk/go/testbus"
)
// 1921/C3: a run ask put on the Go build seat's check stream is pulled by the module started whole on the real bus and
// answered on its step key <node>.<ask id>: the program, run shut off over its heads and its input (asked by digest of the
// seat verb the ask names, which a fake checker answers on the bus as the checker's verb input does), answers its last
// line and exit; a program cannot write under its heads, and one
// that dials the machine's network is shut off, and the answer says it failed.
func TestARunAskIsAnsweredOnTheBusAndItsProgramIsShutOff(t *testing.T) {
needTools(t, "the shut-off run", "bash", "bwrap")
nc, err := nats.Connect(testbus.URL(t))
if err != nil {
t.Fatal(err)
}
defer nc.Close()
js, err := jetstream.New(nc)
if err != nil {
t.Fatal(err)
}
ctx, stop := context.WithCancel(context.Background())
defer stop()
if err := buildstep.DeclareAskWorkers(ctx, js, delivery.AskGo, 2*time.Minute); err != nil {
t.Fatal(err)
}
t.Cleanup(func() { _ = js.DeleteStream(context.Background(), "SEAT_NODE_BUILD_GO") })
tcp, err := net.Listen("tcp", "127.0.0.1:0")
if err != nil {
t.Fatal(err)
}
defer tcp.Close()
states := newStates()
m := moduleWith(states, "go1.26.0")
m.Work, m.Cache, m.Home = t.TempDir(), t.TempDir(), t.TempDir()
m.Capacity = func() int { return 1 }
m.Run = StepRunner(m.Home)
for _, c := range m.caches() {
if err := os.MkdirAll(c, 0o700); err != nil {
t.Fatal(err)
}
}
m.PlaceHead = func(_ context.Context, h delivery.RunHead, dir string) error { return os.MkdirAll(dir, 0o700) }
body := `{"v":1}`
// The checker's verb input on the bus, and the module asking it as the runtime calls a seat verb it invokes.
asked := make(chan string, 4)
sub, err := nc.Subscribe("mesh.seat.mesh-delivery-checker.tool.input", func(msg *nats.Msg) {
var ask delivery.RunInputAsk
_ = json.Unmarshal(msg.Data, &ask)
asked <- ask.Digest
reply, _ := json.Marshal(map[string]any{"result": delivery.RunInputAnswer{Digest: ask.Digest, Input: []byte(body)}})
_ = msg.Respond(reply)
})
if err != nil {
t.Fatal(err)
}
defer sub.Unsubscribe()
m.AskInput = func(key string, b any) (json.RawMessage, error) {
if key != runInputVerb {
return nil, errors.New("asked " + key)
}
data, _ := json.Marshal(b)
msg, err := nc.Request("mesh.seat.mesh-delivery-checker.tool.input", data, 5*time.Second)
if err != nil {
return nil, err
}
var r struct {
Result json.RawMessage `json:"result"`
}
return r.Result, json.Unmarshal(msg.Data, &r)
}
a := runAskOf(t, body)
key := delivery.RunKey(a.Run.Program, delivery.ToolchainUsed{Version: "go1.26.0", Recipe: m.Recipe.Version()}, nil)
exe := m.programPath("programs", key)
os.MkdirAll(filepath.Dir(exe), 0o700)
script := "#!/bin/bash\n" + `if touch "$MESH_RUN_HEADS/novox/mesh-sdk/x" 2>/dev/null; then echo '{"wrote":true}'; exit 9; fi; echo "start"; exec 3<>/dev/tcp/127.0.0.1/` + strings.Split(tcp.Addr().String(), ":")[1] + " || exit 7\n" + `echo '{"dialed":true}'` + "\n"
if err := os.WriteFile(exe, []byte(script), 0o700); err != nil {
t.Fatal(err)
}
p := m.Puller(fakeSteps(new([]string)), &buildstep.JetStreamAsks{JS: js, Language: delivery.AskGo})
p.PullWait = 200 * time.Millisecond
go p.Pull(ctx)
raw, err := delivery.EncodeStepAsk(a)
if err != nil {
t.Fatal(err)
}
subject, _ := delivery.AskSubject(delivery.AskGo, delivery.AskCheck)
if _, err := js.Publish(ctx, subject, raw); err != nil {
t.Fatal(err)
}
stepKey, _ := delivery.AskStepKey("ace", a.ID)
var step delivery.Step
for deadline := time.Now().Add(60 * time.Second); time.Now().Before(deadline); time.Sleep(50 * time.Millisecond) {
states.mu.Lock()
v := states.steps[stepKey]
states.mu.Unlock()
if len(v) > 0 && v[len(v)-1].State != delivery.StepRunning {
step = v[len(v)-1]
break
}
}
if step.State != delivery.StepPassed {
t.Fatalf("the run ask ended %q: %s", step.State, step.Reason)
}
var ans delivery.RunAnswer
if err := json.Unmarshal(step.Result, &ans); err != nil {
t.Fatal(err)
}
if ans.Key != key || ans.Exit != 7 || strings.Contains(string(ans.Output), "dialed") || string(ans.Output) != `"start"` {
t.Fatalf("the answer of a program that dials out = %+v output %s", ans, ans.Output)
}
select {
case d := <-asked:
if d != a.Run.Input.Digest {
t.Errorf("the verb input was asked %s, want %s", d, a.Run.Input.Digest)
}
default:
t.Error("the run's input was not asked of the checker's verb input")
}
}
@@ -0,0 +1,154 @@
package main
import (
"context"
"crypto/sha256"
"encoding/hex"
"encoding/json"
"errors"
"fmt"
"os"
"path/filepath"
"strings"
"testing"
"time"
"git.novox.be/novox/mesh-catalog/modules/mesh-build-go/buildstep"
sdkstep "git.novox.be/novox/mesh-sdk/go/buildstep"
"git.novox.be/novox/mesh-sdk/go/composer"
"git.novox.be/novox/mesh-sdk/go/delivery"
"git.novox.be/novox/mesh-sdk/go/forgekey"
)
const runCommit = "5555555555555555555555555555555555555555"
// runInputVerb is the seat verb a run ask's input is served on, as the checker names it in the ask.
const runInputVerb = "seat:mesh-delivery-checker.input"
// answersInput is a seat verb serving body by its digest, as the checker's verb input answers (delivery.RunInputAnswer).
func answersInput(body string) func(key string, b any) (json.RawMessage, error) {
return func(key string, b any) (json.RawMessage, error) {
ask, _ := b.(delivery.RunInputAsk)
if sum := sha256.Sum256([]byte(body)); key != runInputVerb || ask.Digest != hex.EncodeToString(sum[:]) {
return nil, fmt.Errorf("the checker keeps no input %s", ask.Digest)
}
return json.Marshal(delivery.RunInputAnswer{Digest: ask.Digest, Input: []byte(body)})
}
}
func runAskOf(t *testing.T, body string) delivery.StepAsk {
t.Helper()
in, err := delivery.RunInputOf(runInputVerb, []byte(body))
if err != nil {
t.Fatal(err)
}
return delivery.StepAsk{ID: sdkstep.NewID(), Kind: delivery.AskCheck, Language: delivery.AskGo, Asker: "mesh-delivery-checker", At: time.Now(),
Run: &delivery.RunJob{Program: delivery.RunProgram{Repository: "novox/mesh-delivery", Commit: runCommit, Dir: ".", Package: "./cmd/prog"},
Heads: []delivery.RunHead{{Repository: "novox/mesh-sdk", Commit: runCommit}, {Repository: "novox/mesh-host", Commit: runCommit}},
Input: in, Args: []string{"-seats"}}}
}
// 1921/C1: a kept checking program is run over the input checked against its digest and the heads laid under
// MESH_RUN_HEADS at <owner>/<name>, and the answer carries its exit and last line; nothing is fetched or built; a
// program that fails, or says no JSON, is an answer; an input that is not the one named is refused; and the program was
// kept under delivery.RunKey, so a second ask finds it.
func TestAKeptProgramIsRunOverItsHeadsAndInput(t *testing.T) {
m := composeModule(t)
body := `{"v":1,"case":"x"}`
m.AskInput = answersInput(body)
m.PlaceHead = func(_ context.Context, h delivery.RunHead, dir string) error {
return os.MkdirAll(dir, 0o700)
}
a := runAskOf(t, body)
key := delivery.RunKey(a.Run.Program, delivery.ToolchainUsed{Version: "go1.26.0", Recipe: m.Recipe.Version()}, nil)
exe := m.programPath("programs", key)
os.MkdirAll(filepath.Dir(exe), 0o700)
script := "#!/bin/sh\n" +
`echo "noise"; echo "args $*" >&2; ` +
`printf '{"input":%s,"heads":"%s"}\n' "$(cat "$MESH_RUN_INPUT")" "$(cd "$MESH_RUN_HEADS" && ls -d */* | tr '\n' ' ')"; exit 3` + "\n"
if err := os.WriteFile(exe, []byte(script), 0o700); err != nil {
t.Fatal(err)
}
fetched := 0
fetch := func(context.Context, map[string]any) (any, error) { fetched++; return nil, nil }
out, err := m.run(context.Background(), fetch, a)
if err != nil {
t.Fatal(err)
}
ans := out.(delivery.RunAnswer)
if ans.Built || ans.Key != key || fetched != 0 || ans.Exit != 3 || ans.Said != "args -seats" ||
string(ans.Output) != `{"input":{"v":1,"case":"x"},"heads":"novox/mesh-host novox/mesh-sdk "}` {
t.Fatalf("answer = %+v (fetched %d) output %s", ans, fetched, ans.Output)
}
if entries, _ := os.ReadDir(m.Work); len(entries) != 0 {
t.Errorf("the run's workspace was left: %v", entries)
}
// Shut off, the heads are bound read-only after the writable workspace, so a write under MESH_RUN_HEADS fails: a
// read-only bind before the workspace's is covered by it.
m.Inside = false
var argv []string
m.Run = func(_ context.Context, _ []string, _ string, args ...string) forgekey.Ran {
argv = args
return forgekey.Ran{Status: 0, Stdout: "{}"}
}
if _, err := m.run(context.Background(), fetch, a); err != nil {
t.Fatal(err)
}
line, ws, heads := " "+strings.Join(argv, " ")+" ", -1, -1
for i := 0; i+2 < len(argv); i++ {
switch {
case argv[i] == "--bind" && filepath.Dir(argv[i+1]) == m.Work:
ws = i
case argv[i] == "--ro-bind" && filepath.Base(argv[i+1]) == "heads" && argv[i+2] == argv[i+1]:
heads = i
}
}
if ws < 0 || heads < ws || !strings.Contains(line, " "+delivery.RunHeadsEnv+" "+argv[heads+1]+" ") {
t.Errorf("the heads are not bound read-only over the workspace: %s", line)
}
// An answer over the bus's value limit fails the step, naming its size.
m.Run = func(context.Context, []string, string, ...string) forgekey.Ran {
return forgekey.Ran{Status: 0, Stdout: `"` + strings.Repeat("x", delivery.RunInputBound) + `"`}
}
if _, err := m.run(context.Background(), fetch, a); err == nil || !strings.Contains(err.Error(), "over the bound") {
t.Errorf("an answer over the bound: %v", err)
}
m.Inside, m.Run = true, StepRunner(t.TempDir())
// An input other than the one named is refused, naming it; one the verb does not answer fails the step, saying why.
m.AskInput = func(_ string, b any) (json.RawMessage, error) {
return json.Marshal(delivery.RunInputAnswer{Digest: b.(delivery.RunInputAsk).Digest, Input: []byte(`{"other":1}`)})
}
if _, err := m.run(context.Background(), fetch, a); err == nil || !strings.Contains(err.Error(), "not the one the ask names") || !errors.As(err, new(*buildstep.Refusal)) {
t.Errorf("a wrong input: %v", err)
}
m.AskInput = answersInput(`{"v":2}`)
if _, err := m.run(context.Background(), fetch, a); err == nil || !strings.Contains(err.Error(), "keeps no input") {
t.Errorf("an input the verb does not answer: %v", err)
}
}
// 1921/C2 (F2): a compose ask's seats reach the composer: the module decodes and fills the request and writes it again,
// and used to drop the seats.
func TestAComposeAskKeepsItsSeats(t *testing.T) {
m := composeModule(t)
read, digests := indexerHolds(t, `{"module":"demo","version":"1"}`)
m.ReadManifest = read
a := composeAskOf(t, digests[0])
req := composer.Request{V: composer.Version, Seats: []json.RawMessage{json.RawMessage(`{"seat":"node-x"}`)},
Cases: []composer.Case{{Name: "change", Manifests: map[string]composer.Manifest{"demo": {Digest: digests[0]}}}}}
a.Compose.Request, _ = json.Marshal(req)
key := delivery.ComposerKey("novox/mesh-sdk", composeCommit, "go", composer.Package, delivery.ToolchainUsed{Version: "go1.26.0", Recipe: m.Recipe.Version()}, nil)
exe := m.composerPath(key)
os.MkdirAll(filepath.Dir(exe), 0o700)
script := "#!/bin/sh\n" + `if grep -q '"seats":\[{"seat":"node-x"}\]' "$2"; then S=true; else S=false; fi; ` +
`echo '{"v":1,"sdk":"stand-in","cases":[{"name":"change","manifests":{"demo":{"accepted":'$S'}}}]}'` + "\n"
os.WriteFile(exe, []byte(script), 0o700)
out, err := m.compose(context.Background(), nil, a)
if err != nil {
t.Fatal(err)
}
var v composer.Verdict
if err := json.Unmarshal(out.(delivery.ComposeAnswer).Verdict, &v); err != nil || !v.Cases[0].Manifests["demo"].Accepted {
t.Fatalf("the composer did not receive the seats: %+v %v", v, err)
}
}
@@ -45,6 +45,7 @@ func (m *Module) Tools(steps []StepTool) []stdio.Tool {
tools = append(tools, m.ReadTools()...)
tools = append(tools, buildstep.StepEndTool(StepEndTool, m.Node, m.stepScan(), m.Held))
tools = append(tools, m.releaseTool())
tools = append(tools, m.keptTools()...)
return append(tools,
stdio.Tool{Name: StatusTool,
Description: "What the Go build module is on this machine: the Go version its recipe builds with and the one installed, " +
+5
View File
@@ -23,6 +23,9 @@
"mesh-delivery-checker.kept",
"mesh-delivery-indexer.manifests"
],
"invokes": [
"seat:mesh-delivery-checker.input"
],
"config": {
"capacity": {
"type": "number",
@@ -50,6 +53,8 @@
"build_go_build",
"build_go_imports",
"build_go_release",
"build_go_kept",
"build_go_kept_drop",
"build_go_step",
"build_go_steps",
"build_go_step_end",