Compare commits

..
Author SHA1 Message Date
jschoubben 06cdac97e3 Pin the group's mesh-sdk head 0d97c09c for the retake class (issue 1748)
mesh/delivery-group group fix/1748-retake-and-resolver checking: 2 of 2 member(s) ready
mesh/delivery closed: the pull request closed unmerged
mesh/check pass 06cdac97e32c on fd4931a75261 | judged by the candidate's mesh-build-go at 3200cb79deed
Claude-Session: https://claude.ai/code/session_01JG73KJNrH5EgFXMKKYSrig
2026-11-04 22:45:25 +01:00
jschoubben 82aaae846e mesh-build-go opens a network failure of the fetch with the retake prefix (issue 1748)
The builder takes a retakeable failure again by itself after a backoff.

Claude-Session: https://claude.ai/code/session_01JG73KJNrH5EgFXMKKYSrig
2026-11-04 22:41:57 +01:00
54 changed files with 274 additions and 1804 deletions
-4
View File
@@ -49,10 +49,6 @@ 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
+8 -14
View File
@@ -27,19 +27,16 @@ 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, and the ReadOnlyInside ones (inside the Workspace) read-only over it;
// downloaded) read-only;
// - 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
// 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
Workspace string
Writable []string
ReadOnly []string
Caches []string
WriteCaches bool
Env []string
// Hosts is the file bound as /etc/hosts; Isolated writes one naming only the loopback.
Hosts string
}
@@ -94,7 +91,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}, {"read-only directory", iso.ReadOnlyInside}, {"cache", iso.Caches}} {
}{{"writable directory", iso.Writable}, {"read-only directory", iso.ReadOnly}, {"cache", iso.Caches}} {
for _, d := range group.dirs {
if err := checkDir(group.what, d); err != nil {
return nil, err
@@ -121,9 +118,6 @@ 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")
}
+3 -34
View File
@@ -561,36 +561,6 @@ 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
@@ -606,10 +576,9 @@ 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. 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)
// 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
}
if lacks := delivery.Lacks(ask.Needs, p.Has); len(lacks) > 0 {
@@ -519,39 +519,3 @@ 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,9 +121,6 @@ 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 m.programPath("composers", key)
return filepath.Join(m.Cache, "composers", key, "mesh-compose")
}
// compose runs a compose ask: the composer of its commit, built once, over its request with the manifests filled.
@@ -50,11 +50,7 @@ func (m *Module) compose(ctx context.Context, fetch buildstep.StepFunc, a delive
if err != nil {
return nil, err
}
versions, err := sdkstep.PackagesArg(job.Fetch["packages"])
if err != nil {
return nil, buildstep.Refuse("the compose ask %s: %v", a.ID, err)
}
key := delivery.ComposerKey(repository, commit, job.Dir, job.Package, delivery.ToolchainUsed{Version: installed, Recipe: m.Recipe.Version()}, versions)
key := delivery.ComposerKey(repository, commit, job.Dir, job.Package, delivery.ToolchainUsed{Version: installed, Recipe: m.Recipe.Version()})
exe, built, err := m.composerOf(ctx, fetch, job, key)
if err != nil {
return nil, err
@@ -98,25 +94,17 @@ 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) {
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()
exe := m.composerPath(key)
mu, _ := composerLocks.LoadOrStore(exe, &sync.Mutex{})
mu.(*sync.Mutex).Lock()
defer mu.(*sync.Mutex).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, fetchArgs)
out, err := fetch(ctx, job.Fetch)
if err != nil {
return "", false, err
}
@@ -133,11 +121,11 @@ func (m *Module) programOf(ctx context.Context, fetch buildstep.StepFunc, kind,
defer m.releaseSaid(f.Workspace)
done := m.spaces().Hold(f.Workspace)
defer done()
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)
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)
}
into := filepath.Join(m.Work, f.Workspace, what, "built")
into := filepath.Join(m.Work, f.Workspace, "composer", "mesh-compose")
if err := os.MkdirAll(filepath.Dir(into), 0o700); err != nil {
return "", false, err
}
@@ -145,15 +133,15 @@ func (m *Module) programOf(ctx context.Context, fetch buildstep.StepFunc, kind,
if err != nil {
return "", false, err
}
env := append(m.goEnvFor(tree), "GOPROXY=off", "GOTOOLCHAIN=local", "CGO_ENABLED=0")
r := run(ctx, env, "go", "-C", tree, "build", "-trimpath", "-o", into, pkg)
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)
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 %s %s does not build at %s: %s", what, pkg, fetchArgs["commit"], compileError(r.Err+"\n"+r.Stderr+"\n"+r.Stdout))
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))
}
if err := copyExecutable(into, exe); err != nil {
return "", false, fmt.Errorf("the %s could not be kept: %w", what, err)
return "", false, fmt.Errorf("the composer could not be kept: %w", err)
}
return exe, true, nil
}
@@ -3,7 +3,6 @@ package main
import (
"context"
"encoding/json"
"errors"
"os"
"path/filepath"
"slices"
@@ -65,7 +64,7 @@ func TestAKeptComposerIsRunWithTheManifestsFilled(t *testing.T) {
read, digests := indexerHolds(t, `{"module":"demo","version":"1"}`)
m.ReadManifest = read
a := composeAskOf(t, digests[0])
key := delivery.ComposerKey("novox/mesh-sdk", composeCommit, "go", composer.Package, delivery.ToolchainUsed{Version: "go1.26.0", Recipe: m.Recipe.Version()}, nil)
key := delivery.ComposerKey("novox/mesh-sdk", composeCommit, "go", composer.Package, delivery.ToolchainUsed{Version: "go1.26.0", Recipe: m.Recipe.Version()})
exe := m.composerPath(key)
if err := os.MkdirAll(filepath.Dir(exe), 0o700); err != nil {
t.Fatal(err)
@@ -96,40 +95,13 @@ func TestAKeptComposerIsRunWithTheManifestsFilled(t *testing.T) {
}
}
// 1921: the versions an ask names are an input of the composer it builds, so a composer kept for one set is not run for
// another; an ask naming none keeps the key it always had.
func TestAComposerIsKeptPerAskedVersions(t *testing.T) {
m := composeModule(t)
m.ReadManifest, _ = indexerHolds(t)
a := composeAskOf(t, "feedfeedfeedfeedfeedfeedfeedfeedfeedfeedfeedfeedfeedfeedfeedfeed")
tc := delivery.ToolchainUsed{Version: "go1.26.0", Recipe: m.Recipe.Version()}
plain := delivery.ComposerKey("novox/mesh-sdk", composeCommit, "go", composer.Package, tc, nil)
a.Compose.Fetch["packages"] = map[string]any{"git.novox.be/novox/mesh-sdk/go": strings.Repeat("3", 40)}
named := delivery.ComposerKey("novox/mesh-sdk", composeCommit, "go", composer.Package, tc, delivery.PackageVersions{"git.novox.be/novox/mesh-sdk/go": strings.Repeat("3", 40)})
if plain == named {
t.Fatal("the key ignores the versions")
}
// A composer kept under the key without versions is not used for the ask that names some: it is fetched.
os.MkdirAll(filepath.Dir(m.composerPath(plain)), 0o700)
os.WriteFile(m.composerPath(plain), []byte("#!/bin/sh\nexit 0\n"), 0o700)
fetched := false
fetch := func(context.Context, map[string]any) (any, error) {
fetched = true
return nil, errors.New("stop")
}
m.compose(context.Background(), fetch, a)
if !fetched {
t.Error("a composer built for no versions was run for an ask naming some")
}
}
// 1622: a manifest the indexer does not keep fails the ask, naming its digest; a fetched commit with no go.mod where
// the ask builds fails it naming the directory, and its workspace is released.
func TestAComposeAskFailsLoudly(t *testing.T) {
m := composeModule(t)
m.ReadManifest, _ = indexerHolds(t)
a := composeAskOf(t, "feedfeedfeedfeedfeedfeedfeedfeedfeedfeedfeedfeedfeedfeedfeedfeed")
key := delivery.ComposerKey("novox/mesh-sdk", composeCommit, "go", composer.Package, delivery.ToolchainUsed{Version: "go1.26.0", Recipe: m.Recipe.Version()}, nil)
key := delivery.ComposerKey("novox/mesh-sdk", composeCommit, "go", composer.Package, delivery.ToolchainUsed{Version: "go1.26.0", Recipe: m.Recipe.Version()})
os.MkdirAll(filepath.Dir(m.composerPath(key)), 0o700)
os.WriteFile(m.composerPath(key), []byte("#!/bin/sh\nexit 0\n"), 0o700)
if _, err := m.compose(context.Background(), nil, a); err == nil || !strings.Contains(err.Error(), "feedfeed") {
@@ -150,17 +122,11 @@ func TestAComposeAskFailsLoudly(t *testing.T) {
}
}
// 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).
// 1622: the module reads the delivery indexer's manifests, which a compose ask names by digest.
func TestTheModuleReadsTheIndexersManifests(t *testing.T) {
m := readManifest(t)
if !slices.Contains(m.Reads, ManifestsState) {
if m := readManifest(t); !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
@@ -202,7 +168,7 @@ func TestAComposeAskBuildsTheNodeEnginesComposerAtItsRoot(t *testing.T) {
t.Fatal(err)
}
ans := out.(delivery.ComposeAnswer)
key := delivery.ComposerKey("novox/mesh-host", composeCommit, delivery.RootPath, nodeEngineComposer, delivery.ToolchainUsed{Version: "go1.26.0", Recipe: m.Recipe.Version()}, nil)
key := delivery.ComposerKey("novox/mesh-host", composeCommit, delivery.RootPath, nodeEngineComposer, delivery.ToolchainUsed{Version: "go1.26.0", Recipe: m.Recipe.Version()})
var v composer.Verdict
if err := json.Unmarshal(ans.Verdict, &v); err != nil {
t.Fatal(err)
@@ -19,7 +19,6 @@ import (
"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"
"git.novox.be/novox/mesh-sdk/go/forgekey"
)
@@ -114,7 +113,8 @@ func (m *Module) fetch(ctx context.Context, args map[string]any) (res any, err e
}
}
}()
return m.fetchInto(ctx, req, id, siblings)
res, err = m.fetchInto(ctx, req, id, siblings)
return res, retakeable(err) // a network failure is a retakeable one (issue 1748)
}
func (m *Module) fetchInto(ctx context.Context, req sdkstep.FetchRequest, id string, siblings []siblingArg) (any, error) {
@@ -178,9 +178,7 @@ func (m *Module) fetchInto(ctx context.Context, req sdkstep.FetchRequest, id str
if taken[i], errs[i] = m.takeVersions(ctx, root, filepath.Join(tree, d), d, req.Packages, &pins); errs[i] != nil {
return
}
var more []TakenVersion
more, errs[i] = m.downloadModules(ctx, root, repository, filepath.Join(tree, d), d, &pins, req.Packages)
taken[i] = append(taken[i], more...)
errs[i] = m.downloadModules(ctx, root, filepath.Join(tree, d), d, &pins)
}()
}
wg.Wait()
@@ -197,29 +195,21 @@ func (m *Module) fetchInto(ctx context.Context, req sdkstep.FetchRequest, id str
// downloadModules downloads and verifies the Go modules of the directory dir (d in the tree). A pin of a forge module
// that Go cannot fetch is looked for among the forge's pull request heads and handed to Go by a modfile (pins.go); a
// failure that names no such pin is the first failure. A requirement at the zero version is taken from the asked
// versions (pins.go takeZeroPins).
func (m *Module) downloadModules(ctx context.Context, root, repository, dir, d string, pins *sync.Mutex, asked delivery.PackageVersions) ([]TakenVersion, error) {
var taken []TakenVersion
// failure that names no such pin is the first failure.
func (m *Module) downloadModules(ctx context.Context, root, dir, d string, pins *sync.Mutex) error {
for {
failure := m.downloadAndVerify(ctx, dir, d)
if failure == nil {
return taken, nil
}
// A requirement at the zero version names no commit: it is taken from the asked versions or the trunk's latest.
took, err := m.takeZeroPins(ctx, root, repository, dir, d, failure.Error(), asked, pins)
taken = append(taken, took...)
if err != nil {
return taken, err
return nil
}
pins.Lock()
progress, err := m.resolvePins(ctx, root, dir, failure.Error())
pins.Unlock()
if err != nil {
return taken, err
return err
}
if !progress && len(took) == 0 {
return taken, failure
if !progress {
return failure
}
}
}
@@ -348,7 +338,6 @@ 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.
@@ -1,130 +0,0 @@
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
}},
}
}
@@ -1,70 +0,0 @@
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,7 +58,6 @@ 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,12 +93,7 @@ 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)
// 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
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 == KeptTool || tool.Name == KeptDropTool || tool.Name == StepEndTool {
if tool.Name == StatusTool || tool.Name == HealthTool || tool.Name == StepReadTool || tool.Name == StepsReadTool || tool.Name == ToolchainsReadTool || tool.Name == ReleaseTool || tool.Name == StepEndTool {
continue
}
var reason string
+1 -149
View File
@@ -12,14 +12,10 @@ import (
"context"
"errors"
"fmt"
"io/fs"
"maps"
"os"
"path/filepath"
"regexp"
"slices"
"strings"
"sync"
"git.novox.be/novox/mesh-sdk/go/delivery"
"git.novox.be/novox/mesh-sdk/go/forgekey"
@@ -27,9 +23,6 @@ import (
const forgePrefix = "git.novox.be/novox/"
// zeroShort is the 12 digits a requirement at the zero version v0.0.0-00010101000000-000000000000 ends in: no commit.
const zeroShort = "000000000000"
var pseudoVersion = regexp.MustCompile(`[-.][0-9]{14}-([0-9a-f]{12})$`)
// pin is a requirement of a go.mod on a commit of a repository of the forge, by a pseudo-version.
@@ -115,9 +108,6 @@ func (m *Module) resolvePins(ctx context.Context, root, dir, failure string) (bo
}
progress := false
for _, p := range forgePins(string(raw)) {
if p.short == zeroShort {
continue // takeZeroPins': a requirement at the zero version names no commit to look for
}
if !strings.Contains(failure, p.short) || strings.Contains(done, "replace "+p.module+" =>") {
continue
}
@@ -151,7 +141,7 @@ func (m *Module) placePin(ctx context.Context, root string, p pin, failure strin
r := m.Run(ctx, []string{"GIT_TERMINAL_PROMPT=0"}, "git", "-c", "credential.helper=", "ls-remote", m.remote(repository), "refs/pull/*/head")
if r.Status != 0 || r.Err != "" {
// The forge did not answer: retaken, with its own cause, never held as a refusal until the module changes (issue 1748).
return "", fmt.Errorf("the pin of %s at commit %s: listing the pull requests of %s failed (a download failure that may be retaken): %s", p.module, p.short, repository, forgekey.Said(r.Err+"\n"+r.Stderr))
return "", fmt.Errorf(delivery.Retake+"the pin of %s at commit %s: listing the pull requests of %s failed (a download failure, taken again by the builder): %s", p.module, p.short, repository, forgekey.Said(r.Err+"\n"+r.Stderr))
}
commit, ref := "", ""
for _, l := range strings.Split(r.Stdout, "\n") {
@@ -252,141 +242,3 @@ func absoluteReplaces(body []byte, dir string) []byte {
}
return []byte(strings.Join(lines, "\n"))
}
// takeZeroPins takes, for each requirement of dir's go.mod at the zero version that a go failure names, the version it is
// built against (ADR 0573 §2, §4: whatever go.mod names): the commit the ask names for the package; else the head of
// the package's repository among the asked versions (the check's or try's members: another package of the same
// repository names it); else the latest published version, the trunk's head. It never looks for the commit
// 000000000000 among the pull request heads. It answers what it took; none when no zero requirement is named.
func (m *Module) takeZeroPins(ctx context.Context, root, repository, dir, d, failure string, asked delivery.PackageVersions, pins *sync.Mutex) ([]TakenVersion, error) {
raw, err := os.ReadFile(filepath.Join(dir, "go.mod"))
if err != nil || !strings.Contains(failure, zeroShort) {
return nil, nil
}
resolved := map[string]string{}
if modfile, ok := m.pinModfile(dir); ok {
if b, err := os.ReadFile(modfile); err == nil {
resolved, _ = goModRequirements(string(b))
for module := range resolved {
if strings.Contains(string(b), "replace "+module+" =>") {
resolved[module] = "replaced"
}
}
}
}
var taken []TakenVersion
for _, p := range forgePins(string(raw)) {
if p.short != zeroShort {
continue
}
if v := resolved[p.module]; v != "" && !strings.HasSuffix(v, zeroShort) {
continue // taken before
}
named := "v0.0.0-00010101000000-" + zeroShort
if "novox/"+p.repo == repository {
// A module of the checked repository itself: it is in the fetched tree and nowhere else, since nothing of the
// repository is published before its merge. Every module of the tree goes in, as one may require another.
own, err := m.useOwnModules(dir, root, p.repo)
if err != nil {
return taken, fmt.Errorf("the requirement of %s at the zero version: %w", p.module, err)
}
if !slices.Contains(own, p.module) {
return taken, fmt.Errorf("the requirement of %s at the zero version: no module of that path is in the fetched tree of %s, and nothing of a repository is published before its merge", p.module, repository)
}
tv := TakenVersion{Dir: d, Package: p.module, Named: named, Built: "the fetched tree", Said: "a module of the checked repository: built from its tree"}
if m.Say != nil {
m.Say("%s: %s names no commit; built from the fetched tree of %s", d, p.module, repository)
}
taken = append(taken, tv)
continue
}
tv := TakenVersion{Dir: d, Package: p.module, Asked: askedFor(asked, p), Named: named}
why := "the ask names it"
if tv.Asked == "" {
tv.Asked, why = "latest", "the ask names no commit of its repository, so its latest published version"
} else if _, ok := asked[p.module]; !ok {
why = "the head of its repository among the asked versions"
}
if err := m.takeVersion(ctx, root, dir, &tv, pins); err != nil {
return taken, fmt.Errorf("the requirement of %s at the zero version: %w", p.module, err)
}
tv.Said = strings.TrimPrefix(tv.Said+"; "+why, "; ")
if m.Say != nil {
m.Say("%s: %s names no commit; built at %s (%s)", d, p.module, tv.Built, why)
}
taken = append(taken, tv)
}
return taken, nil
}
// askedFor is the commit the ask names for the package p: its own, else that of the first package of the same repository
// in order (the members' heads are by repository), else none.
func askedFor(asked delivery.PackageVersions, p pin) string {
if c := asked[p.module]; c != "" {
return c
}
for _, pkg := range slices.Sorted(maps.Keys(asked)) {
if pkg == forgePrefix+p.repo || strings.HasPrefix(pkg, forgePrefix+p.repo+"/") {
return asked[pkg]
}
}
return ""
}
// useOwnModules replaces, in dir's temporary modfile, every module of the fetched tree whose path lies in the repository
// repo by its directory in the tree (once), and answers the paths of those modules. A replace of a module dir does not
// require is harmless to Go; one module of a repository requires another of it at the zero version.
func (m *Module) useOwnModules(dir, root, repo string) ([]string, error) {
tree := filepath.Join(root, "tree")
prefix := forgePrefix + repo
type found struct{ module, where string }
var mods []found
err := filepath.WalkDir(tree, func(path string, e fs.DirEntry, err error) error {
if err != nil {
return err
}
if e.IsDir() {
if n := e.Name(); n == ".git" || n == "vendor" || n == "node_modules" {
return filepath.SkipDir
}
return nil
}
if e.Name() != "go.mod" {
return nil
}
raw, err := os.ReadFile(path)
if err != nil {
return err
}
for _, line := range strings.Split(string(raw), "\n") {
if f := strings.Fields(line); len(f) == 2 && f[0] == "module" {
if mod := strings.Trim(f[1], `"`); (mod == prefix || strings.HasPrefix(mod, prefix+"/")) && filepath.Dir(path) != dir {
mods = append(mods, found{mod, filepath.Dir(path)})
}
break
}
}
return nil
})
if err != nil {
return nil, err
}
modfile, _ := m.pinModfile(dir)
done := ""
if modfile != "" {
if b, err := os.ReadFile(modfile); err == nil {
done = string(b)
}
}
var paths []string
for _, f := range mods {
paths = append(paths, f.module)
if strings.Contains(done, "replace "+f.module+" =>") {
continue
}
if err := m.replaceInModfile(dir, f.module, f.where); err != nil {
return nil, err
}
}
return paths, nil
}
@@ -290,112 +290,3 @@ func TestARelativeReplaceStillResolvesInTheTemporaryModfile(t *testing.T) {
t.Fatalf("go mod verify in the pins directory: %v\n%s", err, out)
}
}
// A go.mod requirement at the zero version v0.0.0-00010101000000-000000000000 names no commit (ADR 0573 §2, §4: whatever
// go.mod names): it is taken from the commit the ask names for the package, else from the head of its repository among
// the asked versions, else from the trunk's latest published version, and never looked for as the commit 000000000000
// among the pull request heads.
func TestARequirementAtTheZeroVersionIsTakenFromTheAskOrTheTrunk(t *testing.T) {
const sdk = "git.novox.be/novox/mesh-sdk/go"
zeroMod := "module fixture/member\n\ngo 1.26\n\nrequire (\n\t" + sdk + " v0.0.0-00010101000000-" + zeroShort + "\n)\n"
trunk := strings.Repeat("a", 40)
for _, c := range []struct {
name string
packages map[string]any
download string // what go mod download -json answers
want string // the version asked of go
head bool // the asked commit is on no trunk: only a pull request head carries it
}{
{name: "the ask names the package", packages: map[string]any{sdk: trunk}, want: sdk + "@" + trunk,
download: `{"Version":"v0.1.15-0.20261101000000-aaaaaaaaaaaa","Sum":"h1:zip="}`},
{name: "the ask names another package of its repository", packages: map[string]any{"git.novox.be/novox/mesh-sdk/other": trunk}, want: sdk + "@" + trunk,
download: `{"Version":"v0.1.15-0.20261101000000-aaaaaaaaaaaa","Sum":"h1:zip="}`},
{name: "the ask names the root of its repository, which no package of is published", packages: map[string]any{"git.novox.be/novox/mesh-sdk": trunk}, want: sdk + "@" + trunk,
download: `{"Version":"v0.1.15-0.20261101000000-aaaaaaaaaaaa","Sum":"h1:zip="}`},
{name: "the ask names the head of a pull request", packages: map[string]any{sdk: commitSHA}, want: sdk + "@" + commitSHA, head: true},
{name: "the ask names nothing of its repository", packages: map[string]any{"git.novox.be/novox/mesh-tools/go": trunk}, want: sdk + "@latest",
download: `{"Version":"v0.1.20-0.20261104000000-bbbbbbbbbbbb","Sum":"h1:zip="}`},
{name: "no ask", want: sdk + "@latest", download: `{"Version":"v0.1.20-0.20261104000000-bbbbbbbbbbbb","Sum":"h1:zip="}`},
} {
t.Run(c.name, func(t *testing.T) {
h, states := newPinHarness(t, zeroMod, false, pullList)
base := h.m.Run
h.m.Run = func(ctx context.Context, env []string, name string, args ...string) forgekey.Ran {
line := name + " " + strings.Join(args, " ")
switch {
case name == "go" && strings.Contains(line, " mod download -json "):
h.mu.Lock()
h.cmds = append(h.cmds, cmdRecord{env: env, line: line})
h.mu.Unlock()
if c.head {
return forgekey.Ran{Status: 1, Stdout: `{"Error":"invalid version: unknown revision ` + commitSHA[:12] + `"}`}
}
return forgekey.Ran{Stdout: c.download}
case name == "go" && strings.Contains(line, " mod download") && !strings.Contains(strings.Join(env, " "), "-modfile="):
return forgekey.Ran{Status: 1, Stderr: "go: " + sdk + "@v0.0.0-00010101000000-" + zeroShort + ": invalid version: unknown revision " + zeroShort + "\n"}
}
return base(ctx, env, name, args...)
}
args := map[string]any{"repository": "novox/mesh-delivery", "commit": commitSHA}
if c.packages != nil {
args["packages"] = c.packages
}
if _, err := fetchTool(t, h.m)(args); err != nil {
t.Fatal(err)
}
st := states.waitEnd(t)
if st.State != delivery.StepPassed {
t.Fatalf("the step ended %s: %s", st.State, st.Reason)
}
if !h.ran(" mod download -json " + c.want) {
t.Errorf("go was not asked for %s:\n%s", c.want, strings.Join(h.f.lines(), "\n"))
}
// The pull request heads are listed only to place an asked head, never to look for the commit of zeros.
if listed := h.ran("ls-remote"); listed != c.head {
t.Errorf("the pull request heads were listed: %v, want %v", listed, c.head)
}
if h.ran("pinned/mesh-sdk@" + zeroShort) {
t.Error("a repository was placed at the commit of zeros")
}
})
}
}
// A requirement at the zero version of a module of the checked repository itself (a nested module not yet published, which
// another module of the repository requires) is built from the fetched tree, and never asked of Go or the forge (issue 1921).
func TestARequirementAtTheZeroVersionOfTheCheckedRepositorysOwnModuleIsTakenFromItsTree(t *testing.T) {
const own = "git.novox.be/novox/mesh-delivery/nested/go"
zeroMod := "module fixture/member\n\ngo 1.26\n\nrequire (\n\t" + own + " v0.0.0-00010101000000-" + zeroShort + "\n)\n"
h, states := newPinHarness(t, zeroMod, false, pullList)
base := h.m.Run
h.m.Run = func(ctx context.Context, env []string, name string, args ...string) forgekey.Ran {
line := name + " " + strings.Join(args, " ")
if name == "go" && strings.Contains(line, " mod download") && !strings.Contains(strings.Join(env, " "), "-modfile=") {
trees, _ := filepath.Glob(filepath.Join(h.m.Work, "*", "tree"))
for _, tree := range trees {
os.MkdirAll(filepath.Join(tree, "nested", "go"), 0o755)
os.WriteFile(filepath.Join(tree, "nested", "go", "go.mod"), []byte("module "+own+"\n\ngo 1.26\n"), 0o644)
}
return forgekey.Ran{Status: 1, Stderr: "go: " + own + "@v0.0.0-00010101000000-" + zeroShort + ": invalid version: unknown revision " + zeroShort + "\n"}
}
return base(ctx, env, name, args...)
}
if _, err := fetchTool(t, h.m)(map[string]any{"repository": "novox/mesh-delivery", "commit": commitSHA}); err != nil {
t.Fatal(err)
}
st := states.waitEnd(t)
if st.State != delivery.StepPassed {
t.Fatalf("the step ended %s: %s", st.State, st.Reason)
}
modfiles, _ := filepath.Glob(filepath.Join(h.m.Work, "*", "pins", "*", "go.mod"))
if len(modfiles) != 1 {
t.Fatalf("modfiles = %v", modfiles)
}
body, _ := os.ReadFile(modfiles[0])
if !strings.Contains(string(body), "replace "+own+" => "+filepath.Join(filepath.Dir(filepath.Dir(filepath.Dir(modfiles[0]))), "tree", "nested", "go")) {
t.Errorf("the module of the tree was not replaced into the modfile:\n%s", body)
}
if h.ran(" mod download -json ") || h.ran("ls-remote") {
t.Error("a module of the checked repository was asked of Go or looked for among the pull request heads")
}
}
@@ -0,0 +1,39 @@
package main
import (
"fmt"
"strings"
"git.novox.be/novox/mesh-sdk/go/delivery"
)
// networkWords are what a failed lookup, connection or transfer says in Go's, git's and the resolver's words: a failure of
// the network or of the forge that passes by itself (issue 1748), not of the source built.
var networkWords = []string{
"i/o timeout", "no such host", "could not resolve host", "temporary failure in name resolution", "server misbehaving",
"connection refused", "connection reset", "tls handshake timeout", "network is unreachable", "unexpected eof", "http2: server sent goaway",
}
// networkFailure says whether a failure of a download or a fetch is the network's: its words are one of networkWords and
// it is not go not knowing the revision (unknownRevision, a pin on no head), which waits for another fingerprint.
func networkFailure(failure string) bool {
if unknownRevision(failure) {
return false
}
low := strings.ToLower(failure)
for _, w := range networkWords {
if strings.Contains(low, w) {
return true
}
}
return false
}
// retakeable opens a network failure of the fetch step with delivery.Retake, so the builder takes the key again after a
// backoff (delivery.Retakeable, issue 1748); any other error, one that opens with a prefix already included, is as it was.
func retakeable(err error) error {
if err == nil || strings.HasPrefix(err.Error(), delivery.Retake) || !networkFailure(err.Error()) {
return err
}
return fmt.Errorf("%s%w", delivery.Retake, err)
}
@@ -0,0 +1,42 @@
package main
import (
"errors"
"strings"
"testing"
"git.novox.be/novox/mesh-sdk/go/delivery"
)
// 1748/AC2: a download or a fetch that fails on the network ends with an error that opens with the retake prefix; a
// compile error, a failing test and an unknown revision (a pin on no head) do not.
func TestANetworkFailureOfTheFetchOpensWithTheRetakePrefix(t *testing.T) {
for _, c := range []struct {
name, failure string
retake bool
}{
{"a lookup that times out", "go mod download in .: git.novox.be/novox/mesh-sdk/go@v1: dial tcp: lookup git.novox.be on 10.10.0.3:53: read udp 10.10.0.3:42292->10.10.0.3:53: i/o timeout", true},
{"a refused connection", "git ls-remote -q origin: dial tcp 10.10.0.1:443: connect: connection refused", true},
{"a forge that cannot be resolved", "fatal: unable to access the forge: Could not resolve host: git.novox.be", true},
{"an unknown revision", "invalid version: unknown revision abcdef123456", false},
{"an unknown revision on a cut network is still one go knows", "unknown revision abcdef123456: i/o timeout", false},
{"a compile error", "go build of tool: ./main.go:3:1: undefined: x", false},
{"a failing test", "go test ./...: FAIL: TestX", false},
} {
t.Run(c.name, func(t *testing.T) {
got := retakeable(errors.New(c.failure))
if has := strings.HasPrefix(got.Error(), delivery.Retake); has != c.retake {
t.Fatalf("retakeable(%q) = %q", c.failure, got)
}
if c.retake && (!delivery.Retakeable(got.Error()) || !strings.Contains(got.Error(), c.failure)) {
t.Errorf("%q does not keep the cause or is not Retakeable", got)
}
})
}
if retakeable(nil) != nil {
t.Error("retakeable(nil) is an error")
}
if once := retakeable(errors.New(delivery.Retake + "connection refused")); strings.Count(once.Error(), delivery.Retake) != 1 {
t.Errorf("the prefix was added twice: %q", once)
}
}
@@ -1,132 +0,0 @@
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
}
@@ -1,141 +0,0 @@
//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")
}
}
@@ -1,154 +0,0 @@
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,7 +45,6 @@ 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, " +
@@ -105,10 +105,6 @@ func (m *Module) takeVersion(ctx context.Context, root, dir string, tv *TakenVer
// Not a commit Go can fetch: the head of a group's pull request (ADR 0573 §4), placed from the forge's heads.
// Whether it is one is the forge's fact (placePin), whatever go said.
failure := strings.TrimSpace(info.Error + "\n" + r.Err + "\n" + r.Stderr)
if !commitOK.MatchString(tv.Asked) {
// "latest" (the trunk's head) is no commit of a pull request: there is nothing to place.
return fmt.Errorf("go cannot fetch the latest version of %s, and the ask names no commit of its repository to build it at: either no version of it is published, or the forge did not answer: %s", tv.Package, causeLine(failure))
}
repo, sub, _ := strings.Cut(strings.TrimPrefix(tv.Package, forgePrefix), "/")
p := pin{module: tv.Package, repo: repo, sub: sub, short: tv.Asked[:12]}
pins.Lock()
@@ -117,7 +113,7 @@ func (m *Module) takeVersion(ctx context.Context, root, dir string, tv *TakenVer
pins.Unlock()
// On no head, and go failed for another cause (the forge or the resolver did not answer): it fails with go's own
// words, and not as a kind the builder holds until the module changes (issue 1748).
return fmt.Errorf("downloading the asked version %s of %s failed (a download failure that may be retaken): %s", tv.Asked, tv.Package, causeLine(failure))
return fmt.Errorf(delivery.Retake+"downloading the asked version %s of %s failed (a download failure, taken again by the builder): %s", tv.Asked, tv.Package, causeLine(failure))
}
if err == nil {
err = m.replaceInModfile(dir, tv.Package, where)
@@ -92,7 +92,7 @@ func TestAnAskedVersionIsWrittenIntoTheModfileWhateverGoModNames(t *testing.T) {
st := states.waitEnd(t)
if c.failsWith != "" {
if st.State != delivery.StepFailed || !strings.Contains(st.Reason, c.failsWith) || strings.Contains(st.Reason, "no pull request head") ||
!strings.Contains(st.Reason, "may be retaken") || delivery.FailedUntilChanged(st.Reason) {
!strings.HasPrefix(st.Reason, delivery.Retake) || delivery.FailedUntilChanged(st.Reason) {
t.Errorf("the download failure ended %s: %q", st.State, st.Reason)
}
return
+1 -1
View File
@@ -4,7 +4,7 @@ go 1.26.0
require (
git.novox.be/novox/mesh-catalog/modules/mesh-build-go/buildstep v0.0.0
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261102191058-f02f142af128
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261104214149-0d97c09c49b4
github.com/distribution/distribution/v3 v3.1.2
github.com/nats-io/nats-server/v2 v2.11.17
github.com/nats-io/nats.go v1.54.0
+2 -2
View File
@@ -1,5 +1,5 @@
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261102191058-f02f142af128 h1:aI1DhZAuqVBXbybSqhLNOGX5rUOrn9T0cHNCDUQF7IQ=
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261102191058-f02f142af128/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY=
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261104214149-0d97c09c49b4 h1:K2niszrlnJ7JMiXI6xg4gG90YCd/tqNLGBvujLInI48=
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261104214149-0d97c09c49b4/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY=
github.com/alicebob/miniredis/v2 v2.35.0 h1:QwLphYqCEAo1eu1TqPRN2jgVMPBweeQcR21jeqDCONI=
github.com/alicebob/miniredis/v2 v2.35.0/go.mod h1:TcL7YfarKPGDAthEtl5NBeHZfeUQj6OXMm/+iu5cLMM=
github.com/antithesishq/antithesis-sdk-go v0.7.0-default-no-op h1:Z/MZK75wC/NSrkgqeNIa7jexam9uWzhLmFTSCPI/kn0=
-5
View File
@@ -23,9 +23,6 @@
"mesh-delivery-checker.kept",
"mesh-delivery-indexer.manifests"
],
"invokes": [
"seat:mesh-delivery-checker.input"
],
"config": {
"capacity": {
"type": "number",
@@ -53,8 +50,6 @@
"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",
+1 -3
View File
@@ -1,3 +1 @@
# the binary, built at the module or in its command directory; never the command directory itself
/mesh-build-node
/cmd/mesh-build-node/mesh-build-node
mesh-build-node
@@ -13,7 +13,6 @@ import (
"encoding/json"
"fmt"
"os"
"path"
"path/filepath"
"sort"
"strings"
@@ -38,27 +37,6 @@ type artifact struct {
Language string `json:"language"`
Entrypoints []string `json:"entrypoints"`
External []string `json:"external"`
// Paths are the directories of a package artifact, relative to the module's: one, holding its package.json and
// its package-lock.json (as a Go package's, mesh-build-go gopackage.go); none means the module's directory.
Paths []string `json:"paths"`
}
// packageDir is the directory of a package artifact relative to its module's directory in dir: the one path it names
// when that is a directory of the module holding a package.json (the contracts folders: ts/seats/<seat>/, each with its
// own lock, ADR 0634 section 8), else the module's directory itself (a package that names no path, or source paths
// beside the module's package.json: today's form).
func (a artifact) packageDir(dir string) string {
if len(a.Paths) != 1 {
return "."
}
p := path.Clean(strings.TrimSuffix(a.Paths[0], "/"))
if p == "." || path.IsAbs(p) || p == ".." || strings.HasPrefix(p, "../") {
return "."
}
if _, err := os.Stat(filepath.Join(dir, filepath.FromSlash(p), "package.json")); err != nil {
return "."
}
return p
}
// BuildResult is what a passed build step says.
@@ -110,8 +88,7 @@ func (m *Module) build(ctx context.Context, args map[string]any) (any, error) {
return nil, err
}
if a.Kind == "package" {
at := a.packageDir(dir)
return m.buildPackage(ctx, workspace, module, root, filepath.Join(dir, filepath.FromSlash(at)), at, a, publish)
return m.buildPackage(ctx, workspace, module, root, dir, a, publish)
}
if why, err := lockProblem(dir, module); err != nil || why != "" {
if err == nil {
@@ -16,7 +16,6 @@ import (
"fmt"
"io/fs"
"os"
"path"
"path/filepath"
"strings"
"time"
@@ -100,89 +99,78 @@ func (m *Module) test(ctx context.Context, args map[string]any) (any, error) {
if err != nil {
return nil, err
}
scripts, err := packageScripts(dir)
if err != nil {
return nil, err
}
sources, err := typeScriptFiles(dir)
if err != nil {
return nil, err
}
res := &TestResult{Workspace: workspace, Module: module, Level: level, By: sdkstep.By("mesh-build-node")}
script := "test"
if level == "integration" {
script = "test:integration"
}
hasScript := scripts[script] != ""
if level == "integration" && !hasScript {
res.Passed, res.NoTests, res.Why = true, true, "no tests: the package has no test:integration script"
return res, nil
}
if !hasScript && len(sources) == 0 {
res.Passed, res.NoTests, res.Why = true, true, "no tests: the package has no test script and no .ts files"
return res, nil
}
if err := m.provide(root); err != nil {
return nil, fmt.Errorf("providing the toolchain's packages to the workspace: %w", err)
}
where := module
if where == "" {
where = "."
}
targets, err := testTargets(dir, where)
if err != nil {
return nil, err
}
var check, run time.Duration
var failed []Failure
tested := false
for _, tg := range targets {
scripts, err := packageScripts(tg.dir)
if err != nil {
return nil, err
// a command that could not start is not a verdict: the step fails. One that ran and said no is.
ran := func(name string, r forgekey.Ran, report string) error {
if r.Err != "" {
return fmt.Errorf("%s could not be run: status %d %s: %s", name, r.Status, r.Err, firstLine(r.Stderr))
}
sources, err := typeScriptFiles(tg.dir)
if err != nil {
return nil, err
}
hasScript := scripts[script] != ""
if (level == "integration" && !hasScript) || (!hasScript && len(sources) == 0) {
continue
}
tested = true
if err := m.provide(root); err != nil {
return nil, fmt.Errorf("providing the toolchain's packages to the workspace: %w", err)
}
// a command that could not start is not a verdict: the step fails. One that ran and said no is.
ran := func(name string, r forgekey.Ran, report string) error {
if r.Err != "" {
return fmt.Errorf("%s could not be run: status %d %s: %s", name, r.Status, r.Err, firstLine(r.Stderr))
}
if r.Status != 0 {
failed = append(failed, testFailures(tg.where, name, tg.dir, r, report)...)
}
return nil
}
if len(sources) > 0 {
// A package that builds (a build or prepare script, which the fetch's --ignore-scripts never runs) is emitted
// by tsc into the workspace, shut off like every run step, so its tests find what they import of its own output
// (issue 1869: mesh-sdk's tests import ../dist); any other is type-checked without emitting.
name, line := "type-check", m.typeCheck(tg.dir, sources)
if emit := m.emitBuild(tg.dir, scripts); emit != nil {
name, line = "build", emit
}
start := m.now()
r := m.runIn(ctx, root, false, tg.dir, nil, "node", line...)
check += m.now().Sub(start)
if err := ran(name, r, ""); err != nil {
return nil, err
}
res.Ran = appendOnce(res.Ran, name)
}
if hasScript && len(failed) == 0 {
start := m.now()
// node --test writes its TAP report to a file of the workspace (issue 1901), whatever the script prints itself
report := filepath.Join(root, ".mesh-node-test-"+level+".tap")
_ = os.Remove(report)
env := []string{"PATH=" + binDir(root) + ":" + pathOrDefault(), "npm_config_cache=" + filepath.Join(m.Toolchain.Paths.Cache, "npm"),
"NODE_OPTIONS=" + buildstep.NodeTestOptions(report)}
r := m.runIn(ctx, root, false, tg.dir, env, "npm", scriptArgs(script)...)
run += m.now().Sub(start)
tap, _ := os.ReadFile(report)
_ = os.Remove(report)
if err := ran(script, r, string(tap)); err != nil {
return nil, err
}
res.Ran = appendOnce(res.Ran, script)
if r.Status != 0 {
failed = append(failed, testFailures(where, name, dir, r, report)...)
}
return nil
}
if !tested {
res.Passed, res.NoTests = true, true
res.Why = "no tests: the package has no test script and no .ts files"
if level == "integration" {
res.Why = "no tests: the package has no test:integration script"
if len(sources) > 0 {
// A package that builds (a build or prepare script, which the fetch's --ignore-scripts never runs) is emitted
// by tsc into the workspace, shut off like every run step, so its tests find what they import of its own output
// (issue 1869: mesh-sdk's tests import ../dist); any other is type-checked without emitting.
name, line := "type-check", m.typeCheck(dir, sources)
if emit := m.emitBuild(dir, scripts); emit != nil {
name, line = "build", emit
}
return res, nil
start := m.now()
r := m.runIn(ctx, root, false, dir, nil, "node", line...)
check = m.now().Sub(start)
if err := ran(name, r, ""); err != nil {
return nil, err
}
res.Ran = append(res.Ran, name)
}
if hasScript && len(failed) == 0 {
start := m.now()
// node --test writes its TAP report to a file of the workspace (issue 1901), whatever the script prints itself
report := filepath.Join(root, ".mesh-node-test-"+level+".tap")
_ = os.Remove(report)
env := []string{"PATH=" + binDir(root) + ":" + pathOrDefault(), "npm_config_cache=" + filepath.Join(m.Toolchain.Paths.Cache, "npm"),
"NODE_OPTIONS=" + buildstep.NodeTestOptions(report)}
r := m.runIn(ctx, root, false, dir, env, "npm", scriptArgs(script)...)
run = m.now().Sub(start)
tap, _ := os.ReadFile(report)
_ = os.Remove(report)
if err := ran(script, r, string(tap)); err != nil {
return nil, err
}
res.Ran = append(res.Ran, script)
}
res.Failed = failed
var why []string
@@ -306,56 +294,3 @@ func typeScriptFiles(dir string) ([]string, error) {
})
return out, err
}
// testTarget is a directory whose tests a level runs, and the name it is reported under.
type testTarget struct{ dir, where string }
// testTargets are the directories a module's tests run in: each TypeScript package artifact's own path when every
// package the module declares names one (ADR 0634 §8: its package.json, its lock, its tests), else the module's
// directory.
func testTargets(dir, where string) ([]testTarget, error) {
whole := []testTarget{{dir, where}}
raw, err := os.ReadFile(filepath.Join(dir, "module.json"))
if err != nil {
return whole, nil
}
var mj struct {
Build struct {
Artifacts []artifact `json:"artifacts"`
} `json:"build"`
}
if json.Unmarshal(raw, &mj) != nil {
return whole, nil
}
var out []testTarget
seen := map[string]bool{}
for _, a := range mj.Build.Artifacts {
if a.Language != "typescript" {
continue
}
if a.Kind != "package" {
return whole, nil // a bundle is tested in its module's directory
}
at := a.packageDir(dir)
if at == "." {
return whole, nil
}
if !seen[at] {
seen[at] = true
out = append(out, testTarget{filepath.Join(dir, filepath.FromSlash(at)), path.Join(where, at)})
}
}
if len(out) == 0 {
return whole, nil
}
return out, nil
}
func appendOnce(list []string, s string) []string {
for _, x := range list {
if x == s {
return list
}
}
return append(list, s)
}
@@ -17,7 +17,6 @@ import (
"fmt"
"io/fs"
"os"
"path"
"path/filepath"
"regexp"
"sort"
@@ -211,8 +210,8 @@ func (m *Module) fetchInto(ctx context.Context, req sdkstep.FetchRequest, group
return res, nil
}
// typeScriptModules are the directories of the tree that a typescript artifact of a module.json is built in, relative to
// it, in order: the module's directory for a bundle, the artifact's path for a package.
// typeScriptModules are the directories of the tree whose module.json declares a typescript artifact, relative to
// it, in order.
func (m *Module) typeScriptModules(ctx context.Context, tree string) (dirs []string, packages map[string]bool, err error) {
r := m.Run(ctx, m.baseEnv(), "git", "-C", tree, "ls-files", "--", "module.json", "*/module.json")
if r.Status != 0 || r.Err != "" {
@@ -238,21 +237,13 @@ func (m *Module) typeScriptModules(ctx context.Context, tree string) (dirs []str
if json.Unmarshal(raw, &mj) != nil {
continue // a module.json the build step will refuse by name, not a fetch's business
}
// A bundle is built in its module's directory, a package in its own path (ADR 0634 §8: as at a repository's
// root, each package with its own lock); the first artifact to claim a directory decides it.
for _, a := range mj.Build.Artifacts {
if a.Language != "typescript" {
continue
if a.Language == "typescript" {
d := filepath.ToSlash(filepath.Dir(l))
dirs = append(dirs, d)
packages[d] = a.Kind == "package"
break
}
d := filepath.ToSlash(filepath.Dir(l))
if a.Kind == "package" {
d = path.Join(d, a.packageDir(filepath.Join(tree, filepath.FromSlash(d))))
}
if _, seen := packages[d]; seen {
continue
}
dirs = append(dirs, d)
packages[d] = a.Kind == "package"
}
}
sort.Strings(dirs)
@@ -469,13 +469,11 @@ func fetchCommit(ctx context.Context, run forgekey.Runner, dir, remote, commit,
// (publish) publishes it, from this module's own process with the registry's write credential, which no run step is
// given: a version the registry lacks is published; the same packed file already there is done without a publish;
// another packed file under that version fails `version taken:`. The reference is <name>@<version>, answered only when
// the registry holds this build's file; a candidate answers its digest alone. dir is the package's directory, at its
// place in the module's.
func (m *Module) buildPackage(ctx context.Context, workspace, module, root, dir, at string, a artifact, publish bool) (any, error) {
// rel is the package's directory in the workspace, as the fetch installed it (the shims lie under it).
rel := path.Join(module, at)
if module == "" {
rel = path.Join(".", at)
// the registry holds this build's file; a candidate answers its digest alone.
func (m *Module) buildPackage(ctx context.Context, workspace, module, root, dir string, a artifact, publish bool) (any, error) {
rel := module
if rel == "" {
rel = "."
}
p, err := m.buildAndPack(ctx, root, dir, rel, publish)
if err != nil {
@@ -1,93 +0,0 @@
package main
import (
"context"
"path/filepath"
"strings"
"testing"
)
// A folder module (mesh-contracts' root, a domain folder) declares several package artifacts, each at its own path
// with its own package.json and package-lock.json and none at the module's directory (ADR 0634 section 8).
const folderManifest = `{"module":"c","build":{"artifacts":[` +
`{"name":"ts-a","kind":"package","language":"typescript","paths":["ts/seats/a"]},` +
`{"name":"ts-b","kind":"package","language":"typescript","paths":["ts/seats/b/"]}]}}`
func folderFiles(prefix string) map[string]string {
files := map[string]string{prefix + "module.json": folderManifest}
for _, s := range []string{"a", "b"} {
files[prefix+"ts/seats/"+s+"/package.json"] = `{"name":"@x/` + s + `","version":"0.1.0","scripts":{"build":"tsc","test":"node --test"}}`
files[prefix+"ts/seats/"+s+"/package-lock.json"] = `{"lockfileVersion":3}`
files[prefix+"ts/seats/"+s+"/index.ts"] = "export const v = 1;\n"
}
return files
}
// The fetch installs each package artifact at its path from the lock there, and finds no lock missing at the module.
func TestAFolderModulesPackagesAreInstalledEachAtItsPath(t *testing.T) {
m, f := fetchMachine(t, folderFiles("modules/c/"), nil)
out, err := m.fetch(context.Background(), fetchArgs)
if err != nil {
t.Fatal(err)
}
r := out.(FetchResult)
if len(r.Modules) != 2 || r.Modules[0] != "modules/c/ts/seats/a" || r.Modules[1] != "modules/c/ts/seats/b" || len(r.Installed) != 2 {
t.Errorf("result = %+v, want both package directories found and installed", r)
}
var prefixes []string
for _, c := range npmCalls(f) {
for i, a := range c.args {
if a == "--prefix" {
prefixes = append(prefixes, strings.TrimPrefix(c.args[i+1], filepath.Join(m.Work, r.Workspace, "tree")))
}
}
}
if len(prefixes) != 2 || prefixes[0] != "/modules/c/ts/seats/a" || prefixes[1] != "/modules/c/ts/seats/b" {
t.Errorf("npm ci ran with --prefix %v, want each package's own directory", prefixes)
}
}
// The build runs the package's script and pack in its own path, and the test level runs in each path.
func TestAFolderModulesPackageIsBuiltAndTestedInItsPath(t *testing.T) {
m, f, work := newMachine(t)
workspaceWith(t, work, folderFiles(""))
f.effect = func(c call) {
if c.name == "npm" && len(c.args) > 0 && c.args[0] == "pack" {
for i, a := range c.args {
if a == "--pack-destination" {
write(t, filepath.Join(c.args[i+1], "x.tgz"), "packed", 0o644)
}
}
}
}
if _, err := m.build(context.Background(), map[string]any{"workspace": "w1", "module": ".", "artifact": "ts-b"}); err != nil {
t.Fatal(err)
}
want := filepath.Join(work, "w1", "tree", "ts", "seats", "b")
built := 0
for _, c := range f.calls {
if c.name == "npm" {
built++
if c.dir != want {
t.Errorf("%s ran in %s, want %s", joined(c), c.dir, want)
}
}
}
if built != 2 {
t.Errorf("%d npm commands, want the build and the pack", built)
}
f.calls = nil
res, err := m.test(context.Background(), testArgs("unit"))
if err != nil {
t.Fatal(err)
}
tests := map[string]bool{}
for _, c := range f.calls {
if c.name == "npm" {
tests[c.dir] = true
}
}
if r := res.(*TestResult); !r.Passed || len(tests) != 2 || !tests[want] || !tests[filepath.Join(work, "w1", "tree", "ts", "seats", "a")] {
t.Errorf("unit level = %+v, ran tests in %v, want both package paths", r, tests)
}
}
@@ -7,7 +7,6 @@ import (
"strings"
"sync"
"git.novox.be/novox/mesh-delivery/modules/mesh-delivery-forge-bridge/forgemsg"
sdk "git.novox.be/novox/mesh-sdk/go/delivery"
)
@@ -25,7 +24,7 @@ type fakeMerges struct {
manual bool
failed error
// said are the messages the bridge wrote on FORGE, in order.
said []forgemsg.Forge
said []sdk.Forge
}
func newFakeMerges(f *fakeForge) *fakeMerges {
@@ -77,11 +76,11 @@ func (m *fakeMerges) Put(group string, v sdk.Merge) error {
err = f.refuse
}
f.mu.Unlock()
msg := forgemsg.Forge{Repository: p.Repository, Number: p.Number, Head: p.Head.Head}
msg := sdk.Forge{Repository: p.Repository, Number: p.Number, Head: p.Head.Head}
if err != nil {
msg.Kind, msg.Act, msg.Words = forgemsg.ForgeRefused, "merge", err.Error()
msg.Kind, msg.Act, msg.Words = sdk.ForgeRefused, "merge", err.Error()
} else {
msg.Kind, msg.Merge, msg.Root = forgemsg.ForgeMerged, strings.Repeat("c", 40), "root"
msg.Kind, msg.Merge, msg.Root = sdk.ForgeMerged, strings.Repeat("c", 40), "root"
msg.Trees = map[string]string{"app": "tree"}
}
m.say(msg)
@@ -90,12 +89,12 @@ func (m *fakeMerges) Put(group string, v sdk.Merge) error {
}
// say writes a message on FORGE and has the gate hear it, as the runtime does.
func (m *fakeMerges) say(msg forgemsg.Forge) {
body, err := forgemsg.EncodeForge(msg)
func (m *fakeMerges) say(msg sdk.Forge) {
body, err := sdk.EncodeForge(msg)
if err != nil {
panic(err)
}
event, err := forgemsg.ForgeEvent(msg.Repository, msg.Kind)
event, err := sdk.ForgeEvent(msg.Repository, msg.Kind)
if err != nil {
panic(err)
}
@@ -16,7 +16,6 @@ import (
"testing"
"time"
"git.novox.be/novox/mesh-delivery/modules/mesh-delivery-forge-bridge/forgemsg"
stdio "git.novox.be/novox/mesh-sdk/go"
"git.novox.be/novox/mesh-sdk/go/asks"
sdk "git.novox.be/novox/mesh-sdk/go/delivery"
@@ -583,7 +582,7 @@ func TestAPassingVerdictAsksTheApproveAndItsApproveMergesWithinSecondsOverARealB
t.Fatalf("the gate asked the forge to merge %d times: the bridge merges", n)
}
// The bridge says it merged; the mesh's source then holds the merge, in one write.
b.forgeSays(t, forgemsg.Forge{Kind: forgemsg.ForgeMerged, Repository: "novox/app", Number: 7, Head: head, Merge: strings.Repeat("b", 40),
b.forgeSays(t, sdk.Forge{Kind: sdk.ForgeMerged, Repository: "novox/app", Number: 7, Head: head, Merge: strings.Repeat("b", 40),
Root: "r", Trees: map[string]string{"app": "t"}})
// The forge's own word that the pull request merged follows the bridge's (its webhook): the delivery ends merged.
b.publish(t, "mesh.mod.gitea.event.pull.merged", merged("novox/app", 7, head, strings.Repeat("b", 40)))
@@ -678,13 +677,13 @@ func (b *testBus) answer(t *testing.T, key, option string) time.Time {
}
// forgeSays is the forge bridge's message on FORGE, as its stream carries it.
func (b *testBus) forgeSays(t *testing.T, f forgemsg.Forge) {
func (b *testBus) forgeSays(t *testing.T, f sdk.Forge) {
t.Helper()
raw, err := forgemsg.EncodeForge(f)
raw, err := sdk.EncodeForge(f)
if err != nil {
t.Fatal(err)
}
event, err := forgemsg.ForgeEvent(f.Repository, f.Kind)
event, err := sdk.ForgeEvent(f.Repository, f.Kind)
if err != nil {
t.Fatal(err)
}
@@ -743,7 +742,7 @@ func TestAGroupOfTwoWithGreenChecksGetsOneAskAndOnItsApproveBothMergeOneAfterThe
t.Fatalf("the second member was handed before the first merged: %+v", again.Pulls)
}
one := first.Pulls[0]
b.forgeSays(t, forgemsg.Forge{Kind: forgemsg.ForgeMerged, Repository: one.Repository, Number: one.Number, Head: one.Head.Head,
b.forgeSays(t, sdk.Forge{Kind: sdk.ForgeMerged, Repository: one.Repository, Number: one.Number, Head: one.Head.Head,
Merge: strings.Repeat("a", 40), Root: "r", Trees: map[string]string{"x": "t"}})
var both sdk.Merge
rt.until("the second member to be handed", eventWithin, func() bool { both = mergeAt(b, "feat/x"); return len(both.Pulls) == 2 })
@@ -751,7 +750,7 @@ func TestAGroupOfTwoWithGreenChecksGetsOneAskAndOnItsApproveBothMergeOneAfterThe
t.Fatalf("the second merge handed is %+v", both.Pulls)
}
two := both.Pulls[1]
b.forgeSays(t, forgemsg.Forge{Kind: forgemsg.ForgeMerged, Repository: two.Repository, Number: two.Number, Head: two.Head.Head,
b.forgeSays(t, sdk.Forge{Kind: sdk.ForgeMerged, Repository: two.Repository, Number: two.Number, Head: two.Head.Head,
Merge: strings.Repeat("b", 40), Root: "r", Trees: map[string]string{"x": "t"}})
rt.until("source.main to hold both merges", eventWithin, func() bool {
e, err := b.kvs["source"].Get(ctx, sdk.SourceMainKey)
@@ -6,7 +6,7 @@ import (
"testing"
"time"
"git.novox.be/novox/mesh-delivery/modules/mesh-delivery-forge-bridge/forgemsg"
sdk "git.novox.be/novox/mesh-sdk/go/delivery"
)
// The holder, end to end over fakes (novox/hq ADR 0239 *How it is checked*, cut by ADR 0539 §17): a pull request's head
@@ -241,9 +241,9 @@ func TestAMergedGroupEndsMergedWithSourceMainMoved(t *testing.T) {
t.Fatalf("source.main is %+v, wanted both merges", w.source.main)
}
// A merge that leaves no module.json carries no trees (issue 1367): the gate reads it, and refuses nothing.
msg := forgemsg.Forge{Kind: forgemsg.ForgeMerged, Repository: "novox/a", Number: 9, Head: strings.Repeat("e", 40),
msg := sdk.Forge{Kind: sdk.ForgeMerged, Repository: "novox/a", Number: 9, Head: strings.Repeat("e", 40),
Merge: strings.Repeat("f", 40), Root: "r"}
body, err := forgemsg.EncodeForge(msg)
body, err := sdk.EncodeForge(msg)
if err != nil {
t.Fatalf("a merged message with no trees is refused: %v", err)
}
@@ -251,17 +251,6 @@ func TestAMergedGroupEndsMergedWithSourceMainMoved(t *testing.T) {
if len(w.h.refusals) != 0 {
t.Errorf("the gate refused a merged message with no trees: %v", w.h.refusals)
}
// A group's end is the indexer's: the gate ignores it before decoding, and says no refusal (1585/T25).
w.h.Forged("mesh-delivery-forge-bridge.forge.group.merged", []byte(`{"v":1,"group":"feat/y"}`))
if len(w.h.refusals) != 0 {
t.Errorf("the gate refused a group's end: %v", w.h.refusals)
}
// The repository forge/group's merged message ends the same way: it is decoded as a forge message, and this one,
// which is none, is refused.
w.h.Forged("mesh-delivery-forge-bridge.forge.forge.group.merged", []byte(`{"v":1,"group":"feat/y"}`))
if len(w.h.refusals) != 1 {
t.Errorf("forge/group's merged message was passed by as a group's end: refusals %v", w.h.refusals)
}
}
// ADR 0419 §2, issue 978: a description's `after:` line is not read, and says so once on its pull request. The
@@ -8,7 +8,6 @@ import (
"testing"
"time"
"git.novox.be/novox/mesh-delivery/modules/mesh-delivery-forge-bridge/forgemsg"
"git.novox.be/novox/mesh-sdk/go/asks"
sdk "git.novox.be/novox/mesh-sdk/go/delivery"
)
@@ -766,6 +765,6 @@ func allGreenFor(w *world, ms ...member) {
// bridgeMerged is the forge bridge's word that it merged a member.
func bridgeMerged(w *world, repository string, number int, head, merge string) {
w.merges.say(forgemsg.Forge{Kind: forgemsg.ForgeMerged, Repository: repository, Number: number, Head: head, Merge: merge, Root: "root",
w.merges.say(sdk.Forge{Kind: sdk.ForgeMerged, Repository: repository, Number: number, Head: head, Merge: merge, Root: "root",
Trees: map[string]string{"app": "tree"}})
}
@@ -5,7 +5,6 @@ import (
"testing"
"time"
"git.novox.be/novox/mesh-delivery/modules/mesh-delivery-forge-bridge/forgemsg"
"git.novox.be/novox/mesh-sdk/go/asks"
sdk "git.novox.be/novox/mesh-sdk/go/delivery"
)
@@ -101,7 +100,7 @@ func TestAnAnswerHandsTheFirstMemberOnlyAndTheBridgesMergedHandsTheNext(t *testi
if len(w.merges.puts) != 1 {
t.Fatalf("the next member was handed before the bridge said the first merged: %d merges written", len(w.merges.puts))
}
w.merges.say(forgemsg.Forge{Kind: forgemsg.ForgeMerged, Repository: gCat, Number: 2, Head: gHk, Merge: strings.Repeat("a", 40), Root: "r", Trees: map[string]string{"x": "t"}})
w.merges.say(sdk.Forge{Kind: sdk.ForgeMerged, Repository: gCat, Number: 2, Head: gHk, Merge: strings.Repeat("a", 40), Root: "r", Trees: map[string]string{"x": "t"}})
w.settleAll()
if len(w.merges.puts) != 2 {
t.Fatalf("the second member was not handed after the first merged: %d merges written", len(w.merges.puts))
@@ -116,7 +115,7 @@ func TestAnAnswerHandsTheFirstMemberOnlyAndTheBridgesMergedHandsTheNext(t *testi
// AC2: the bridge's refused hands no member, and the group is said urgent, naming the member and the forge's words.
func TestTheBridgesRefusalHandsNoMemberAndIsSaidUrgent(t *testing.T) {
w := answeredFirst(t)
w.merges.say(forgemsg.Forge{Kind: forgemsg.ForgeRefused, Repository: gCat, Number: 2, Head: gHk, Act: "merge",
w.merges.say(sdk.Forge{Kind: sdk.ForgeRefused, Repository: gCat, Number: 2, Head: gHk, Act: "merge",
Words: "the forge refused gitea_merge_pull_request: mesh/repo-check (missing)"})
w.later(time30s)
w.settleAll()
@@ -138,7 +137,7 @@ func TestTheBridgesRefusalHandsNoMemberAndIsSaidUrgent(t *testing.T) {
// the group, even after the first member merged.
func TestTheNextMemberIsNotHandedWhileARequiredStatusOfTheGroupIsNotASuccess(t *testing.T) {
w := answeredFirst(t)
w.merges.say(forgemsg.Forge{Kind: forgemsg.ForgeMerged, Repository: gCat, Number: 2, Head: gHk, Merge: strings.Repeat("a", 40), Root: "r", Trees: map[string]string{"x": "t"}})
w.merges.say(sdk.Forge{Kind: sdk.ForgeMerged, Repository: gCat, Number: 2, Head: gHk, Merge: strings.Repeat("a", 40), Root: "r", Trees: map[string]string{"x": "t"}})
setChecks(w, gCtl, gHc, "success", "") // the second member's repository check went missing meanwhile
w.settleAll()
if len(w.merges.puts) != 1 {
@@ -194,7 +193,7 @@ func TestARefusalThatRepeatsWithNoTrunkMoveIsNotAskedAgainWithoutEnd(t *testing.
w := answeredFirst(t)
d := w.h.deliveries[IDOf(gCat, gHk)]
refuse := func(took string) {
w.h.ForgeRefused(forgemsg.Forge{Kind: forgemsg.ForgeRefused, Repository: gCat, Number: 2, Head: gHk, Act: "merge",
w.h.ForgeRefused(sdk.Forge{Kind: sdk.ForgeRefused, Repository: gCat, Number: 2, Head: gHk, Act: "merge",
Words: "novox/mesh-catalog#2 is not merged: its verdict names no trunk commit it was judged against, and main is now f70a99dbf485. " +
"The check of head d2d2d2d2d2d2 merged into trunk f70a99dbf485 was asked; the last check took " + took + ". Ask again once both statuses pass"})
}
@@ -10,7 +10,6 @@ import (
"strings"
"time"
"git.novox.be/novox/mesh-delivery/modules/mesh-delivery-forge-bridge/forgemsg"
stdio "git.novox.be/novox/mesh-sdk/go"
sdk "git.novox.be/novox/mesh-sdk/go/delivery"
)
@@ -313,33 +312,23 @@ func (h *Holder) mergeValue(name, key string, approved map[string]string, target
return m, len(m.Pulls) > 0 && state == sdk.MergeMerged
}
// forgeGroupEnd is the whole event of a group's end on FORGE, the forge bridge's event name under its module: compared
// exactly, so the merged message of a repository named forge/group, which ends the same way, is read as one.
const forgeGroupEnd = sdk.ForgeBridgeModule + "." + forgemsg.ForgeGroupEndEvent
// Forged takes a message of the forge bridge on FORGE: the gate follows a merge and the refusal of a merge, and nothing else.
//
// The end of a group's merges (`forge.group.merged`, a body this gate has no use for: the gate wrote the merges itself) is
// the indexer's, and is ignored before it is decoded, so no refusal is said for it (issue 1585, ADR 0633 clause 4).
func (h *Holder) Forged(event string, body []byte) {
if event == forgeGroupEnd {
return
}
f, err := forgemsg.DecodeForge(body)
f, err := sdk.DecodeForge(body)
if err != nil {
h.Refuse(fmt.Errorf("a message of the forge bridge is not readable: %v", err))
return
}
switch {
case f.Kind == forgemsg.ForgeMerged:
case f.Kind == sdk.ForgeMerged:
h.ForgeMerged(f)
case f.Kind == forgemsg.ForgeRefused && f.Act == "merge":
case f.Kind == sdk.ForgeRefused && f.Act == "merge":
h.ForgeRefused(f)
}
}
// memberOfForge is the delivery a forge message is about: at the head it names.
func (h *Holder) memberOfForge(f forgemsg.Forge) *Delivery {
func (h *Holder) memberOfForge(f sdk.Forge) *Delivery {
if d := h.deliveries[IDOf(f.Repository, f.Head)]; d != nil && d.Number == f.Number {
return d
}
@@ -348,7 +337,7 @@ func (h *Holder) memberOfForge(f forgemsg.Forge) *Delivery {
// ForgeMerged takes the bridge's word that a pull request the gate handed to it merged: the merge is announced once,
// with the ask, its asker the teller and the heads (issue 1025), and the next member's turn comes at the next look.
func (h *Holder) ForgeMerged(f forgemsg.Forge) {
func (h *Holder) ForgeMerged(f sdk.Forge) {
h.mu.Lock()
defer h.mu.Unlock()
d := h.memberOfForge(f)
@@ -365,7 +354,7 @@ func (h *Holder) ForgeMerged(f forgemsg.Forge) {
// ForgeRefused takes the bridge's word that the forge refused a merge: nothing after it is merged, and the refusal is said
// as a stalled line (ADR 0441 §3, §4). The merge is handed again only at the next event that decides whether it may
// merge (mergeAgainNow).
func (h *Holder) ForgeRefused(f forgemsg.Forge) {
func (h *Holder) ForgeRefused(f sdk.Forge) {
h.mu.Lock()
defer h.mu.Unlock()
d := h.memberOfForge(f)
+1 -2
View File
@@ -3,8 +3,7 @@ module mesh-delivery-gate
go 1.26.0
require (
git.novox.be/novox/mesh-delivery/modules/mesh-delivery-forge-bridge/forgemsg v0.0.0-20261104235927-c78941e80b59
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261104002338-5a38a1221659
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261103174208-b7abe203a929
github.com/nats-io/nats-server/v2 v2.11.17
github.com/nats-io/nats.go v1.54.0
)
+2 -4
View File
@@ -1,7 +1,5 @@
git.novox.be/novox/mesh-delivery/modules/mesh-delivery-forge-bridge/forgemsg v0.0.0-20261104235927-c78941e80b59 h1:pFb+lP0FxS3bF6N3gHKtjzyixI9b1vC/NLViT/lVzRQ=
git.novox.be/novox/mesh-delivery/modules/mesh-delivery-forge-bridge/forgemsg v0.0.0-20261104235927-c78941e80b59/go.mod h1:Xxg9VI4y755tznAS2PygQFe2V3/aIzjRn3/06+Br9nQ=
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261104002338-5a38a1221659 h1:tK7MzxVm/ECNg1Yblhh1ynVEthbZCg2mSPW0wQZC0b0=
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261104002338-5a38a1221659/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY=
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261103174208-b7abe203a929 h1:RtAFbUJRP1qO9AIMM2ue+cXNzqj7WZ7IOlSqNvttwuU=
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261103174208-b7abe203a929/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY=
github.com/antithesishq/antithesis-sdk-go v0.7.0-default-no-op h1:Z/MZK75wC/NSrkgqeNIa7jexam9uWzhLmFTSCPI/kn0=
github.com/antithesishq/antithesis-sdk-go v0.7.0-default-no-op/go.mod h1:FQyySiasQQM8735Ddel3MRojmy4dA1IqCeyJ5jmPMbI=
github.com/google/go-tpm v0.9.8 h1:slArAR9Ft+1ybZu0lBwpSmpwhRXaa85hWtMinMyRAWo=
+1 -1
View File
@@ -16,7 +16,7 @@ operator, it is accepted only on the operator's Approve, given on a channel that
| Verb | Takes | Answers |
|---|---|---|
| `ask` | `question` (at most 500 characters), `bound` | the records and clauses bearing on the question, ranked; each clause says whether it holds, what changed it, or that it is only drafted; each record says what changed it since and how to read its reasoning |
| `get` | `number`, `kind`, `reasoning`, `section`, `version`, `bound` | one record: clauses, links both ways, what changed it since, versions, reviews, the pending version's reviews with their findings (`pending-reviews`), the operator's ask; its reasoning only when asked |
| `get` | `number`, `kind`, `reasoning`, `section`, `version`, `bound` | one record: clauses, links both ways, what changed it since, versions, reviews, the operator's ask; its reasoning only when asked |
| `list` | `kind`, `status`, `topic`, `flag` (`research-done` or `proposed-old`), `before` (a number, the cursor), `limit` | short rows, newest first, number by number: number, kind, title, status, topic, date, versions, and whether a proposal is pending; `next`, the `before` that reads on, when more match |
| `findings` | `number` (optional) | the whole-set findings kept — each a rule's failure, the records it names, what it says, and when; with `number`, only the findings naming that record |
| `holds` | `topic` or `number`, `bound` | the clauses in force for a topic of the reading order or for one record; never a drafted clause |
@@ -1,8 +1,6 @@
package main
import (
"context"
"fmt"
"strings"
"testing"
"time"
@@ -120,33 +118,6 @@ func TestAChangesVerdictLeavesTheServedVersionServedAndGraduatingNeedsABecameLin
e.Current.Status != StatusActive || e.Current.Section("01-findings.md") != "" {
t.Fatalf("after changes: pending %v, current %+v", e.Pending != nil, e.Current.Sections)
}
// 1970/AC1: get with no version shows the pending version's findings, and says where the version stands.
g, err := w.k.Get(GetAsk{Kind: KindResearch, Number: r.Number})
if err != nil {
t.Fatal(err)
}
if len(g.PendingReviews) != 1 || g.PendingReviews[0].Verdict != VerdictChanges || g.PendingReviews[0].Version == 0 ||
!strings.Contains(g.PendingReviews[0].Findings, "say where the finding comes from") || len(g.Reviews) != 1 ||
!strings.Contains(g.Pending, "changes asked") || !strings.Contains(g.Pending, "pending-reviews") {
t.Fatalf("pending findings not shown: %+v, pending %q, reviews %+v", g.PendingReviews, g.Pending, g.Reviews)
}
// 1970/AC3: under a small bound the last pending review is kept, and the cut is said.
e2 := mustGet(t, w.k.Store, key)
more := e2.Pending.Clone()
for i := 0; i < 3; i++ {
more.Reviews = append(more.Reviews, Review{By: "r", Verdict: VerdictChanges, Version: 9, Findings: strings.Repeat("a long finding ", 60) + fmt.Sprint(i)})
}
if err := w.k.Store.Save(context.Background(), Version{By: "author@g14", Why: "test", Record: more}, false, false, e2.Revision); err != nil {
t.Fatal(err)
}
g, err = w.k.Get(GetAsk{Kind: KindResearch, Number: r.Number, Bound: LeastBound})
if err != nil {
t.Fatal(err)
}
if len(g.PendingReviews) != 1 || !strings.HasSuffix(g.PendingReviews[0].Findings, "2") ||
!strings.Contains(g.Cut, "earlier pending review") {
t.Fatalf("pending cut: %d kept, cut %q", len(g.PendingReviews), g.Cut)
}
grad := effortAsk(r.Number, StatusGraduated, overview)
if _, err := w.k.Propose("author@g14", grad); err == nil || !strings.Contains(err.Error(), "became") {
t.Fatalf("graduated without a became link: %v", err)
@@ -312,17 +312,14 @@ type Got struct {
Idea string `json:"idea,omitempty"`
Waiting string `json:"waiting,omitempty"`
// Retracted are the retractions the operator approved; Retracting, one waiting for the operator.
Retracted []RetractionRow `json:"retracted,omitempty"`
Retracting string `json:"retracting,omitempty"`
Reviews []Review `json:"reviews,omitempty"`
// PendingReviews are the reviews of the pending version, with their findings, when get is asked without a
// version (issue 1970): the version waits for its author, and these are what it must answer.
PendingReviews []Review `json:"pending-reviews,omitempty"`
Drafting *Drafting `json:"drafting,omitempty"`
Review string `json:"review,omitempty"`
Source string `json:"source,omitempty"`
Reserved string `json:"reserved,omitempty"`
NeedsOperat bool `json:"needs-operator,omitempty"`
Retracted []RetractionRow `json:"retracted,omitempty"`
Retracting string `json:"retracting,omitempty"`
Reviews []Review `json:"reviews,omitempty"`
Drafting *Drafting `json:"drafting,omitempty"`
Review string `json:"review,omitempty"`
Source string `json:"source,omitempty"`
Reserved string `json:"reserved,omitempty"`
NeedsOperat bool `json:"needs-operator,omitempty"`
// Brief is what the operator's ask shows of it (issue 639).
Brief *Brief `json:"brief,omitempty"`
Cut string `json:"cut,omitempty"`
@@ -374,17 +371,7 @@ func (k *KB) Get(a GetAsk) (*Got, error) {
g.Idea = "an idea: research it again before anything rests on it"
}
if e.Pending != nil && a.Version == 0 {
g.PendingReviews = e.Pending.Reviews
g.Pending = fmt.Sprintf("version %d, a progressive insight or a new version of an active effort, waits for its review: history shows what it changes", e.PendingVersion)
if n := len(e.Pending.Reviews); n > 0 && e.Pending.Reviews[n-1].Verdict == VerdictChanges {
last := e.Pending.Reviews[n-1]
k := len(strings.Split(strings.TrimSpace(last.Findings), "\n"))
if strings.TrimSpace(last.Findings) == "" {
k = 0
}
g.Pending = fmt.Sprintf("version %d: changes asked (%d finding(s)) on version %d, shown in pending-reviews; it waits for its author's revision",
e.PendingVersion, k, last.Version)
}
}
if r.Ask != nil {
g.Waiting = "the operator's ask: " + r.Ask.Said
@@ -465,10 +452,6 @@ func (k *KB) Get(a GetAsk) (*Got, error) {
cut = append(cut, fmt.Sprintf("%d earlier review(s) are not shown: history lists every version", len(g.Reviews)-1))
g.Reviews = g.Reviews[len(g.Reviews)-1:]
}
if size(g) > room && len(g.PendingReviews) > 1 {
cut = append(cut, fmt.Sprintf("%d earlier pending review(s) are not shown: get with the pending version's number shows them", len(g.PendingReviews)-1))
g.PendingReviews = g.PendingReviews[len(g.PendingReviews)-1:]
}
for _, which := range []string{"linked-from", "links"} {
if size(g) <= room {
break
@@ -783,7 +783,7 @@ func Tools(k *KB, caller func(map[string]any) string) []stdio.Tool {
}},
{Name: Seat + ".get",
Description: "One record by its number: its title, status, topic, date and deciders; its clauses, each with whether it holds; " +
"its links both ways; what changed it since; its versions; its reviews, and the findings of the pending version's reviews (pending-reviews) when one waits for its author; the operator's ask when it waits for one. " +
"its links both ways; what changed it since; its versions; its reviews, and the operator's ask when it waits for one. " +
"Its reasoning (context, options, consequences, how it is checked; a research effort's documents) only with reasoning, " +
"or one section by name; an earlier version with version. Within a bound it states, 4 KiB unless asked for more, " +
"saying what it cut.",
@@ -361,15 +361,6 @@ func TestAnAcceptedRecordChangesOnlyByAnInsightReviewedLikeAProposal(t *testing.
e.Pending.Reviews[0].Verdict != VerdictChanges {
t.Fatalf("the insight was served before its review approved it: %+v", e.Current.Sections)
}
// 1970/AC2: the same for a pending insight.
g, err := w.k.Get(GetAsk{Kind: KindDecision, Number: n})
if err != nil {
t.Fatal(err)
}
if len(g.PendingReviews) != 1 || !strings.Contains(g.PendingReviews[0].Findings, "what a minute is measured from") ||
!strings.Contains(g.Pending, "changes asked") {
t.Fatalf("pending insight findings not shown: %+v, %q", g.PendingReviews, g.Pending)
}
w.judge.review = ""
p.Sections[2].Text = "Merges land sooner, by about a minute from the last merge."
if _, err := w.k.Propose("author@g14", p); err != nil {
+2 -13
View File
@@ -302,20 +302,9 @@ func readPlanRaw(key string) ([]byte, error) {
// assignmentKeys is every key of the controller's assignments mirror, read the same way.
func assignmentKeys() ([]string, error) { return stdio.State(AssignmentsRead).Keys() }
// Checked by TestTheVaultWholeRotatesANodesLoginWithAStandInEngine: the race it met (a pass before Serve, every watch
// refused, the rotation waiting for the minute's pass) is what this closes.
// afterServed runs f once served is closed, that is once stdio.Serve holds the runtime's channel: every function of the
// bus refuses before that ("before Serve"), and a watch refused then waits for the minute's pass.
func afterServed(served <-chan struct{}, f func()) {
go func() {
<-served
f()
}()
}
func main() {
env := build()
afterServed(stdio.Served(), func() {
go func() {
tick := time.NewTicker(time.Minute)
defer tick.Stop()
// At start, every covered account it holds whose hash is not of its current generation (ADR 0547 clause 9).
@@ -372,7 +361,7 @@ func main() {
}
<-tick.C
}
})
}()
if err := stdio.Serve("", tools(env)); err != nil {
say("%v", err)
os.Exit(1)
-73
View File
@@ -1416,76 +1416,3 @@ func (h *Holder) AsksNow() []AskView {
})
return out
}
// AsksQuery narrows what the tool `operator-channel.asks` answers (issue 595): Key one ask, State open (open and held) or
// ended, Since the asks opened at or after that time, Bound the most bytes of the answer's asks. The zero query keeps the
// whole answer, as before.
type AsksQuery struct {
Key, State string
Since time.Time
Bound int
}
// AsksAnswer is the answer of a narrowed query: the asks kept, how many the filters matched, and, when the bound cut the
// answer short, that it did and how.
type AsksAnswer struct {
Asks []AskView `json:"asks"`
Matched int `json:"matched"`
Cut string `json:"cut,omitempty"`
}
// AsksQueryOf reads the arguments of the tool; an empty set is the zero query.
func AsksQueryOf(a map[string]any) (AsksQuery, error) {
q := AsksQuery{Key: strArg(a, "key"), State: strArg(a, "state")}
if q.State != "" && q.State != "open" && q.State != "ended" {
return q, errorf("state is open or ended, not %q", q.State)
}
if since := strArg(a, "since"); since != "" {
t, err := time.Parse(time.RFC3339, since)
if err != nil {
return q, errorf("since is a time like 2026-11-04T22:00:00Z: %v", err)
}
q.Since = t
}
if v, ok := a["bound"].(float64); ok {
if v < 1 {
return q, errorf("bound is the most bytes of the answer, at least 1")
}
q.Bound = int(v)
}
return q, nil
}
// Narrow keeps of the views (open first, newest first) the asks the query names, and cuts the answer at its byte bound: the
// asks that fit, in order, and a said cut when one more did not.
func (q AsksQuery) Narrow(all []AskView) AsksAnswer {
var out AsksAnswer
out.Asks = []AskView{}
var kept []AskView
for _, v := range all {
open := v.State == StateOpen || v.State == StateHeld
switch {
case q.Key != "" && v.Key != q.Key,
q.State == "open" && !open,
q.State == "ended" && open,
!q.Since.IsZero() && v.Opened < q.Since.UTC().Format(time.RFC3339):
continue
}
kept = append(kept, v)
}
out.Matched = len(kept)
used := 0
for i, v := range kept {
if q.Bound > 0 {
raw, _ := json.Marshal(v)
if used+len(raw)+1 > q.Bound {
out.Cut = fmt.Sprintf("the answer is cut at its bound of %d bytes: %d of %d asks are shown, the %d after them are not "+
"(narrow with key, state or since, or raise bound)", q.Bound, i, len(kept), len(kept)-i)
break
}
used += len(raw) + 1
}
out.Asks = append(out.Asks, v)
}
return out
}
@@ -2,7 +2,6 @@ package main
import (
"encoding/json"
"reflect"
"strings"
"testing"
"time"
@@ -141,55 +140,3 @@ func TestAHeldAskNamesEveryKindAndWhyWhenNoChannelIsReady(t *testing.T) {
t.Fatalf("%q", why)
}
}
// Issue 595: operator-channel.asks answers one ask, the open or the ended, those opened from a time, and no more bytes than a
// bound, said when it cut the answer short; with no argument it keeps the whole answer.
func TestTheAsksVerbNarrowsItsAnswer(t *testing.T) {
view := func(key, state, opened string) AskView {
return AskView{Key: key, Asker: "mesh-delivery", Headline: "Approve " + key, State: state, Opened: opened}
}
all := []AskView{view("a.3", StateOpen, "2026-11-04T10:00:00Z"), view("a.2", StateHeld, "2026-11-03T10:00:00Z"),
view("a.1", "ended", "2026-11-02T10:00:00Z"), view("a.0", "ended", "2026-11-01T10:00:00Z")}
one, _ := json.Marshal(all[0])
for _, tc := range []struct {
name string
args map[string]any
want []string
matched int
cut bool
err bool
}{
{"no argument is the zero query", map[string]any{}, nil, 0, false, false},
{"key", map[string]any{"key": "a.1"}, []string{"a.1"}, 1, false, false},
{"state open takes open and held", map[string]any{"state": "open"}, []string{"a.3", "a.2"}, 2, false, false},
{"state ended", map[string]any{"state": "ended"}, []string{"a.1", "a.0"}, 2, false, false},
{"since", map[string]any{"since": "2026-11-02T12:00:00Z"}, []string{"a.3", "a.2"}, 2, false, false},
{"bound cuts and says so", map[string]any{"bound": float64(2*len(one) + 10), "state": "open"}, []string{"a.3", "a.2"}, 2, false, false},
{"a bound under the asks cuts", map[string]any{"bound": float64(len(one) + 5)}, []string{"a.3"}, 4, true, false},
{"a state that is neither", map[string]any{"state": "done"}, nil, 0, false, true},
{"a time that is not one", map[string]any{"since": "yesterday"}, nil, 0, false, true},
} {
q, err := AsksQueryOf(tc.args)
if (err != nil) != tc.err {
t.Errorf("%s: err = %v", tc.name, err)
continue
}
if tc.err {
continue
}
if tc.want == nil {
if q != (AsksQuery{}) {
t.Errorf("%s: %+v is not the zero query, which keeps today's answer", tc.name, q)
}
continue
}
got := q.Narrow(all)
var keys []string
for _, v := range got.Asks {
keys = append(keys, v.Key)
}
if !reflect.DeepEqual(keys, tc.want) || got.Matched != tc.matched || (got.Cut != "") != tc.cut {
t.Errorf("%s: asks %v matched %d cut %q, want %v, %d, cut %v", tc.name, keys, got.Matched, got.Cut, tc.want, tc.matched, tc.cut)
}
}
}
+2 -18
View File
@@ -734,24 +734,8 @@ func tools(h *Holder, l *listening) []stdio.Tool {
{Name: "operator-channel.asks",
Description: "The asks this router holds (novox/hq ADR 0259): open (or held, with since when and why no channel can carry it yet) first, then the last month's — each " +
"one's asker, headline, state, when it opened and expires, the channels it was offered on, how it " +
"ended (the warrant: who chose what, through which channel), and whether its asker was told. With no argument every " +
"ask, which is large; filter it: key for one ask, state open or ended, since (a time, RFC 3339) for the asks opened " +
"from then, and bound, the most bytes of the answer, which says when it cut it short.",
Input: map[string]any{
"key": str("one ask, by its key (asker.id); none for every ask"),
"state": str("open (open and held) or ended; none for both"),
"since": str("only the asks opened at or after this time, RFC 3339"),
"bound": map[string]any{"type": "integer", "description": "the most bytes of the asks in the answer; said when it cuts the answer short"}},
Run: func(a map[string]any) (any, error) {
q, err := AsksQueryOf(a)
if err != nil {
return nil, err
}
if q == (AsksQuery{}) {
return h.AsksNow(), nil
}
return q.Narrow(h.AsksNow()), nil
}},
"ended (the warrant: who chose what, through which channel), and whether its asker was told.",
Run: func(map[string]any) (any, error) { return h.AsksNow(), nil }},
{Name: "messenger_identities",
Description: "The accounts linked as the operator, one per channel kind: the account, the name its " +
"service shows, when it was linked, and from when it can approve (an hour after linking).",
+3 -5
View File
@@ -105,11 +105,9 @@ type Machine struct {
// ownerOf reads a file's owner from its inode; nil is the inode as the kernel says it. Tests set it
// so the folders they make stand for a non-root owner, whoever runs them.
ownerOf func(info os.FileInfo) (uid, gid int)
// Beside runs a look below a share beside the pass; nil is a goroutine. Tests run it in place.
Beside func(func())
// looks is the last look below each real path (LookKept).
looks map[string]*lookBelow
looksMu sync.Mutex
// clean is when a look below each real path last found no backup repository (LookKept).
clean map[string]time.Time
cleanMu sync.Mutex
// walk is the running pass's context, which ends the look below a share when the process stops.
walk context.Context
walkMu sync.Mutex
@@ -32,7 +32,7 @@ func newFake(t *testing.T) *fake {
}
func (f *fake) machine() *Machine {
return &Machine{Root: f.root, UID: f.uid, owners: f.owners, ownerOf: asTheLibrarysOwner, Beside: func(f func()) { f() }, Now: func() time.Time { return f.now }, Run: func(_ context.Context, name string, args ...string) Ran {
return &Machine{Root: f.root, UID: f.uid, owners: f.owners, ownerOf: asTheLibrarysOwner, Now: func() time.Time { return f.now }, Run: func(_ context.Context, name string, args ...string) Ran {
line := strings.Join(append([]string{name}, args...), " ")
f.ran = append(f.ran, line)
if (line == "exportfs -ra" || line == "sudo -n exportfs -ra") && f.onExportfs != nil {
+2 -1
View File
@@ -61,8 +61,9 @@ func (m *Machine) Pass(ctx context.Context, last *Status) Status {
if _, note := m.Backup(); note != "" {
s.Warnings = append(s.Warnings, note)
}
m.setWalk(ctx) // stays set: a look beside the pass ends when the process does
m.setWalk(ctx)
exports, refused, said := m.Judge(c, roster, wants)
m.setWalk(nil)
if ctx.Err() != nil {
// Stopping: write nothing and reload nothing, so Stopped starts at once.
s.Problems = append(s.Problems, "the pass was ended by the process stopping; nothing was written")
@@ -432,89 +432,3 @@ func TestAHangingCommandStillLetsStoppedRemoveTheFileInTime(t *testing.T) {
t.Errorf("the export file stays after an unassignment whose commands hung")
}
}
// Issue 1971: the look below a share runs beside the pass. A pass whose walk has not finished still writes
// its status, refuses the share until a look has finished, keeps believing a clean look while it is renewed,
// and a walk ended by the process stopping leaves no result.
func TestTheLookBelowRunsBesideThePassAndNeverHoldsItsStatus(t *testing.T) {
f := newFake(t)
f.mountinfo("/")
f.dir("/srv/clean/films/one")
f.file(ConfigFile, grantsConfig("clean=/srv/clean", "clean=ws:rw"))
f.roster("ws 10.0.0.3")
f.wants(map[string][]Want{"ws": {{Share: "clean", Access: "rw"}}})
m := f.machine()
var walks []func()
m.Beside = func(w func()) { walks = append(walks, w) } // the walk is still running
now := time.Date(2026, 11, 5, 5, 0, 0, 0, time.UTC)
m.Now = func() time.Time { return now }
s := m.Pass(context.Background(), nil)
if !s.At.Equal(now) || f.read(StatusFile) == "" || len(walks) != 1 {
t.Fatalf("the pass waited for the walk: at %v, walks %d", s.At, len(walks))
}
if !strings.Contains(refusedWhy(s, "clean"), "not been looked through yet") {
t.Errorf("a share whose look is unfinished is exported: %+v", s.Refused)
}
m.Pass(context.Background(), &s)
if len(walks) != 1 {
t.Errorf("a second walk started while one runs: %d", len(walks))
}
walks[0]()
if s = m.Pass(context.Background(), &s); refusedWhy(s, "clean") != "" {
t.Fatalf("the finished look was not used: %+v", s.Refused)
}
now = now.Add(LookKept)
s = m.Pass(context.Background(), &s)
if len(walks) != 2 || refusedWhy(s, "clean") != "" {
t.Errorf("the renewal must run beside the pass and the clean look stand: walks %d, %+v", len(walks), s.Refused)
}
ctx, cancel := context.WithCancel(context.Background())
m.Pass(ctx, &s)
cancel()
f.restic("/srv/clean/films/one/backup")
walks[1]()
if s = m.Pass(context.Background(), &s); refusedWhy(s, "clean") != "" {
t.Errorf("a walk ended by the stop left a result: %+v", s.Refused)
}
walks[len(walks)-1]()
if s = m.Pass(context.Background(), &s); !strings.Contains(refusedWhy(s, "clean"), "/srv/clean/films/one/backup") {
t.Errorf("the repository was not found by the next look: %+v", s.Refused)
}
}
// After a restart no look is remembered; a share the previous process's export file carries stays exported,
// and nothing is reloaded, while its first look runs (a delivery never unexports a mounted share).
func TestARestartKeepsTheExportedShareWhileItsFirstLookRuns(t *testing.T) {
f := newFake(t)
f.mountinfo("/", "/srv/clean")
f.dir("/srv/clean/films/one")
f.file(ConfigFile, grantsConfig("clean=/srv/clean", "clean=ws:rw"))
f.roster("ws 10.0.0.3")
f.wants(map[string][]Want{"ws": {{Share: "clean", Access: "rw"}}})
f.kernelReads()
first := f.machine()
s := first.Pass(context.Background(), nil)
before := f.read(ExportFile)
if refusedWhy(s, "clean") != "" || !strings.Contains(before, "/srv/clean ") {
t.Fatalf("the share was not exported to begin with: %+v %q", s.Refused, before)
}
m := f.machine()
var held []func()
m.Beside = func(w func()) { held = append(held, w) }
f.ran = nil
s = m.Pass(context.Background(), &s)
if len(held) != 1 || refusedWhy(s, "clean") != "" || f.read(ExportFile) != before || countExportfs(f.ran) != 0 {
t.Errorf("a restart with an unfinished walk unexported the share: %+v, ran %v", s.Refused, f.ran)
}
// The same name, a new folder holding a repository, the walk unfinished: refused, not carried by name.
f.restic("/srv/other/backup")
f.file(ConfigFile, grantsConfig("clean=/srv/other", "clean=ws:rw"))
m3 := f.machine()
m3.Beside = func(func()) {}
s = m3.Pass(context.Background(), &s)
if refusedWhy(s, "clean") == "" || strings.Contains(f.read(ExportFile), "/srv/other") {
t.Errorf("a share moved to a new folder was exported unlooked: %+v", s.Refused)
}
}
+24 -80
View File
@@ -244,72 +244,33 @@ const (
)
// LookKept is how long a look below a share that found no backup repository is believed. The look walks up
// to WalkLimit folders, which on a large library takes minutes on a cold disk; a repository put under a
// to WalkLimit folders, which on a large library is too much for every 30 s pass; a repository put under a
// share is found within this long. A look that found one, or was cut short, is never kept: the share stays
// refused and is looked at again on the next pass.
const LookKept = time.Hour
// lookBelow is the last look below one folder, and whether another runs now.
type lookBelow struct {
found string
complete bool
at time.Time
done bool // a look has finished; until then there is no result
running bool
}
// repositoryBelow answers the last look below dir for a backup repository, and starts the next one beside
// the caller when there is none, or it was not clean, or LookKept has passed. The pass therefore never waits
// for the walk (issue 1971: a walk of minutes held the status's time back until the module was found
// unhealthy). pending is true while no look has finished yet; the share is refused until then, and while a
// kept clean look is being renewed the old result stands. complete is false when the look stopped at its
// bound or with the process.
func (m *Machine) repositoryBelow(dir string) (found string, complete, pending bool) {
m.looksMu.Lock()
r := m.looks[dir]
if r == nil {
if m.looks == nil {
m.looks = map[string]*lookBelow{}
}
r = &lookBelow{}
m.looks[dir] = r
}
fresh := r.done && r.found == "" && r.complete && m.clock().Sub(r.at) < LookKept
start := !fresh && !r.running
if start {
r.running = true
}
m.looksMu.Unlock()
if start {
walk := func() {
f, c := m.walkBelow(dir)
m.looksMu.Lock()
defer m.looksMu.Unlock()
r.running = false
if w := m.walking(); w != nil && w.Err() != nil {
return // ended by the process stopping: no result, the old one stands
}
r.found, r.complete, r.at, r.done = f, c, m.clock(), true
}
if m.Beside != nil {
m.Beside(walk)
} else {
go walk()
}
}
m.looksMu.Lock()
defer m.looksMu.Unlock()
if !r.done {
return "", false, true
}
return r.found, r.complete, false
}
func (m *Machine) clock() time.Time {
// repositoryBelow finds a backup repository within WalkDepth levels below dir. complete is false when
// the look stopped at its bound before it saw everything.
func (m *Machine) repositoryBelow(dir string) (found string, complete bool) {
now := time.Now()
if m.Now != nil {
return m.Now()
now = m.Now()
}
return time.Now()
m.cleanMu.Lock()
defer m.cleanMu.Unlock()
if at, kept := m.clean[dir]; kept && now.Sub(at) < LookKept {
return "", true
}
found, complete = m.walkBelow(dir)
if found == "" && complete {
if m.clean == nil {
m.clean = map[string]time.Time{}
}
m.clean[dir] = now
} else {
delete(m.clean, dir)
}
return found, complete
}
// walkBelow is the look itself, breadth first, bounded by WalkDepth and WalkLimit.
@@ -381,8 +342,7 @@ func (m *Machine) Judge(c Config, roster map[string]string, wants map[string][]W
prefix, _ = netip.ParsePrefix(c.Range)
}
points := m.MountPoints()
onDisk, _ := m.Read(ExportFile) // what the previous process judged
backup, _ := m.Backup() // judged on every pass, never kept: see backup.go
backup, _ := m.Backup() // judged on every pass, never kept: see backup.go
backup.Paths = m.realPaths(backup.Paths)
shares := map[string]bool{}
for _, s := range c.Shares {
@@ -451,16 +411,12 @@ func (m *Machine) Judge(c Config, roster map[string]string, wants map[string][]W
no(fmt.Sprintf("The share %s's folder %s%s is inside the backup repository %s, and backups are never exported.", s.Name, real, via, inside))
continue
}
holds, complete, pending := m.repositoryBelow(real)
holds, complete := m.repositoryBelow(real)
if holds != "" {
no(fmt.Sprintf("The share %s's folder %s%s holds a backup repository (%s), and backups are never exported: share the folders beside it instead.", s.Name, real, via, holds))
continue
}
if pending && !carries(onDisk, real) {
no(fmt.Sprintf("The share %s's folder %s%s has not been looked through yet for a backup repository (the look runs beside the pass and takes a while on a large folder), so it is not exported yet.", s.Name, real, via))
continue
}
if !pending && !complete {
if !complete {
no(fmt.Sprintf("The share %s's folder %s%s holds more folders than the module looks through (%d levels, at most %d folders), so it could not make sure no backup repository is under it: share smaller folders.", s.Name, real, via, WalkDepth, m.limit()))
continue
}
@@ -531,18 +487,6 @@ func (m *Machine) Judge(c Config, roster map[string]string, wants map[string][]W
return out, refused, said
}
// carries says whether the export file already holds the line for this real folder (a share exported to no
// node has none and waits for its look; a share whose folder changed is a new folder). While a share's first look is unfinished (after a restart) it stays as the previous process judged
// it, so a delivery never unexports a share clients have mounted.
func carries(file, real string) bool {
for _, l := range strings.Split(file, "\n") {
if strings.HasPrefix(l, real+" ") {
return true
}
}
return false
}
func (m *Machine) limit() int {
if m.WalkLimit > 0 {
return m.WalkLimit