Merge the trunk (B0) into the registry group: one invokes list in mesh-build-go's manifest, the manifest asked of its owner and the checker's input asked of the seat (issue 1921, issue 1966)

Claude-Session: https://claude.ai/code/session_01NcbgfFSnn24M3GcqttoU7i
This commit is contained in:
2026-11-07 02:18:55 +01:00
34 changed files with 1008 additions and 80 deletions
+4
View File
@@ -49,6 +49,10 @@ already gone is no error, one a running step holds is refused); its caller calls
failed fetch removes its own workspace. A workspace no caller released is removed two hours after its last use (ADR
0509), and the removal is said in the log as the caller's fault.
The programs this module builds from a commit and keeps in its cache (composers of compose asks, checking programs of run
asks) are read by `build_go_kept` (kind, key, size, last use) and removed by `build_go_kept_drop {kind, key}`; one no ask used
for 30 days is swept with the workspaces (issue 1921).
A pin on a pull request's head (issue 1148): when `go mod download` fails on a `git.novox.be/novox/*` pseudo-version, the
fetch looks for that commit among the forge's `refs/pull/*/head`, places the repository at it under the workspace's
`pinned/beside/` as a plain directory, and hands it to Go (the fetch's commands, and the build with `GOPROXY=off`) through a
+14 -8
View File
@@ -27,16 +27,19 @@ const Bubblewrap = "bwrap"
// - the system's programs and libraries read-only, and of /etc only what a program needs to run (users, libraries,
// certificates, time zone), with a hosts file naming only the loopback;
// - the Workspace and Writable directories read-write, the ReadOnly ones (the toolchain, the modules the fetch step
// downloaded) read-only;
// downloaded) read-only, and the ReadOnlyInside ones (inside the Workspace) read-only over it;
// - the shared warm Caches read-only under a throwaway writable layer (S18), or read-write for a trunk build alone
// (WriteCaches), so a candidate's code never changes what a later build reads.
type Isolation struct {
Workspace string
Writable []string
ReadOnly []string
Caches []string
WriteCaches bool
Env []string
Workspace string
Writable []string
ReadOnly []string
// ReadOnlyInside are directories inside the Workspace or a Writable one bound read-only after them, so they stay
// read-only (a later bind of a path wins; a ReadOnly one inside the Workspace would be covered by it).
ReadOnlyInside []string
Caches []string
WriteCaches bool
Env []string
// Hosts is the file bound as /etc/hosts; Isolated writes one naming only the loopback.
Hosts string
}
@@ -91,7 +94,7 @@ func (iso Isolation) command(env []string, name string, args ...string) ([]strin
for _, group := range []struct {
what string
dirs []string
}{{"writable directory", iso.Writable}, {"read-only directory", iso.ReadOnly}, {"cache", iso.Caches}} {
}{{"writable directory", iso.Writable}, {"read-only directory", iso.ReadOnly}, {"read-only directory", iso.ReadOnlyInside}, {"cache", iso.Caches}} {
for _, d := range group.dirs {
if err := checkDir(group.what, d); err != nil {
return nil, err
@@ -118,6 +121,9 @@ func (iso Isolation) command(env []string, name string, args ...string) ([]strin
for _, d := range iso.Writable {
argv = append(argv, "--bind", d, d)
}
for _, d := range iso.ReadOnlyInside {
argv = append(argv, "--ro-bind", d, d)
}
if iso.Hosts != "" {
argv = append(argv, "--ro-bind", iso.Hosts, "/etc/hosts")
}
+34 -3
View File
@@ -561,6 +561,36 @@ func (p *Puller) declineRun(msg *Pulled, ask sdkstep.StepAsk, key string, step s
p.mu.Unlock()
}
// refuseUnread ends an ask this module cannot read: when its data holds an id that keys a step, the step is written
// refused with the reason, so the asker learns at once that no node began it; otherwise (no id, not JSON) the ask is only
// dropped and logged. The message is dropped either way.
func (p *Puller) refuseUnread(msg *Pulled, why error) {
var probe struct {
ID string `json:"id"`
Kind string `json:"kind"`
}
if json.Unmarshal(msg.Data, &probe) == nil && probe.ID != "" {
if key, kerr := delivery.AskStepKey(p.Node, probe.ID); kerr == nil {
tool := "ask"
if probe.Kind != "" {
tool = probe.Kind + " ask"
}
at := now().UTC()
step := delivery.Step{ID: probe.ID, Tool: tool, Node: p.Node, State: delivery.StepRefused, Started: at, Ended: at,
Reason: fmt.Sprintf("this build module on %s cannot read the ask: %v", p.Node, why), Owner: sdkstep.Owner}
if werr := writeEnd(p.Put, key, step); werr != nil {
p.logf("the unread ask %s cannot be answered on %s: %v", probe.ID, p.Node, werr)
} else {
p.logf("an ask refused: %v", why)
_ = msg.Term()
return
}
}
}
p.logf("an ask dropped: %v", why)
_ = msg.Term()
}
func withKind(msg *Pulled, kind string) *Pulled {
if msg != nil {
msg.Kind = kind
@@ -576,9 +606,10 @@ func (p *Puller) take(ctx context.Context, msg *Pulled) {
err = fmt.Errorf("the ask %s is a %s ask of %s, pulled from the %s %s worker", ask.ID, ask.Kind, ask.Language, p.Language, kind)
}
if err != nil {
// An ask that is not one runs nowhere: delivering it again would only fail again.
p.logf("an ask dropped: %v", err)
_ = msg.Term()
// An ask that is not one runs nowhere: delivering it again would only fail again. But an ask that names an id
// is answered on its step key as refused, naming why (a job this build module does not know, a version it does
// not read): a dropped ask must always be answered, or its asker waits to its bound for nothing.
p.refuseUnread(msg, err)
return
}
if lacks := sdkstep.Lacks(ask.Needs, p.Has); len(lacks) > 0 {
@@ -519,3 +519,39 @@ func TestAPoisonedAskIsSaidAndRunNoMore(t *testing.T) {
t.Errorf("acked %v naked %v, want it termed alone", acked, naked)
}
}
// An ask whose job this build module does not know is answered on its step key as refused, naming why, never only dropped
// (a dropped ask leaves its asker waiting to its bound); one with no id is still dropped.
func TestPullAnswersAnAskItCannotRead(t *testing.T) {
src, st := NewFakeAsks(), newSteps()
good := testAsk(delivery.AskCheck)
raw, err := delivery.EncodeStepAsk(good)
if err != nil {
t.Fatal(err)
}
var m map[string]any
_ = json.Unmarshal(raw, &m)
delete(m, "check")
m["frobnicate"] = map[string]any{"program": "x"}
odd, _ := json.Marshal(m)
// The job is not one this reader holds, so the ask carries none it knows: decode refuses it.
if _, err := delivery.DecodeStepAsk(odd); err == nil {
t.Fatal("the fixture ask decodes")
}
src.Publish(delivery.AskCheck, odd)
src.Publish(delivery.AskCheck, []byte(`{"v":1}`))
p := &Puller{Source: src, Language: delivery.AskGo, Node: "ace", Put: st.put, Capacity: func() int { return 1 },
PullWait: 10 * time.Millisecond, Run: func(ctx context.Context, a delivery.StepAsk) (any, error) {
t.Error("an unread ask ran")
return nil, nil
}}
ctx, cancel := context.WithCancel(context.Background())
defer cancel()
go p.Pull(ctx)
waitFor(t, "both termed", func() bool { _, tm, _ := src.Fates(); return len(tm) == 2 })
key, _ := delivery.AskStepKey("ace", good.ID)
s, ok := st.get(key)
if !ok || s.State != delivery.StepRefused || !strings.Contains(s.Reason, "cannot read the ask") || !strings.Contains(s.Reason, "exactly one") {
t.Fatalf("the unread ask's step = %+v (%v)", s, ok)
}
}
@@ -125,6 +125,9 @@ func (m *Module) Puller(steps []StepTool, source buildstep.AskSource) *buildstep
if a.Compose != nil {
return m.compose(ctx, ask.Fetch, a)
}
if a.Run != nil {
return m.run(ctx, ask.Fetch, a)
}
return buildstep.RunAsk(ctx, ask, a)
}}
}
@@ -46,7 +46,7 @@ var composerLocks sync.Map
// composerPath is where the composer of key is kept: the cache directory, beside Go's own caches.
func (m *Module) composerPath(key string) string {
return filepath.Join(m.Cache, "composers", key, "mesh-compose")
return m.programPath("composers", key)
}
// compose runs a compose ask: the composer of its commit, built once, over its request with the manifests filled.
@@ -129,21 +129,29 @@ 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 sdkstep.ComposeJob, key string) (string, bool, error) {
exe := m.composerPath(key)
mu, _ := composerLocks.LoadOrStore(exe, &sync.Mutex{})
mu.(*sync.Mutex).Lock()
defer mu.(*sync.Mutex).Unlock()
args, err := buildstep.FetchArgs(job.Fetch)
if err != nil {
return "", false, err
}
return m.programOf(ctx, fetch, "composers", "composer", key, args, job.Dir, job.Package)
}
// programOf is the program kept under key in the cache directory kind, or, when none is, the one built now: the commit
// fetch names is fetched (the fetch step: the forge key and the network, none of the change's code), and package in the
// Go module of dir is built in a run step shut off with nothing fetched. what names it in a failure.
func (m *Module) programOf(ctx context.Context, fetch buildstep.StepFunc, kind, what, key string, fetchArgs map[string]any, dir, pkg string) (string, bool, error) {
exe := m.programPath(kind, key)
mu := lockOf(exe)
mu.Lock()
defer mu.Unlock()
if st, err := os.Stat(exe); err == nil && st.Mode().IsRegular() {
touch(exe, m.clock())
return exe, false, nil
}
if fetch == nil {
return "", false, buildstep.Refuse("this build module fetches nothing")
}
args, err := buildstep.FetchArgs(job.Fetch)
if err != nil {
return "", false, err
}
out, err := fetch(ctx, args)
out, err := fetch(ctx, fetchArgs)
if err != nil {
return "", false, err
}
@@ -160,11 +168,11 @@ func (m *Module) composerOf(ctx context.Context, fetch buildstep.StepFunc, job s
defer m.releaseSaid(f.Workspace)
done := m.spaces().Hold(f.Workspace)
defer done()
dir := filepath.Join(m.Work, f.Workspace, "tree", job.Dir)
if _, err := os.Stat(filepath.Join(dir, "go.mod")); err != nil {
return "", false, fmt.Errorf("%s at %s holds no go.mod in %s, so no composer can be built there", job.Fetch.Repository, job.Fetch.Commit, job.Dir)
tree := filepath.Join(m.Work, f.Workspace, "tree", dir)
if _, err := os.Stat(filepath.Join(tree, "go.mod")); err != nil {
return "", false, fmt.Errorf("%s at %s holds no go.mod in %s, so no %s can be built there", fetchArgs["repository"], fetchArgs["commit"], dir, what)
}
into := filepath.Join(m.Work, f.Workspace, "composer", "mesh-compose")
into := filepath.Join(m.Work, f.Workspace, what, "built")
if err := os.MkdirAll(filepath.Dir(into), 0o700); err != nil {
return "", false, err
}
@@ -172,15 +180,15 @@ func (m *Module) composerOf(ctx context.Context, fetch buildstep.StepFunc, job s
if err != nil {
return "", false, err
}
env := append(m.goEnvFor(dir), "GOPROXY=off", "GOTOOLCHAIN=local", "CGO_ENABLED=0")
r := run(ctx, env, "go", "-C", dir, "build", "-trimpath", "-o", into, job.Package)
env := append(m.goEnvFor(tree), "GOPROXY=off", "GOTOOLCHAIN=local", "CGO_ENABLED=0")
r := run(ctx, env, "go", "-C", tree, "build", "-trimpath", "-o", into, pkg)
if r.Status != 0 || r.Err != "" {
// The words Go says when the module holds no such package (a mesh-sdk from before the composer) are kept whole,
// so the asker tells that apart from a fault.
return "", false, fmt.Errorf("the composer %s does not build at %s: %s", job.Package, job.Fetch.Commit, compileError(r.Err+"\n"+r.Stderr+"\n"+r.Stdout))
return "", false, fmt.Errorf("the %s %s does not build at %s: %s", what, pkg, fetchArgs["commit"], compileError(r.Err+"\n"+r.Stderr+"\n"+r.Stdout))
}
if err := copyExecutable(into, exe); err != nil {
return "", false, fmt.Errorf("the composer could not be kept: %w", err)
return "", false, fmt.Errorf("the %s could not be kept: %w", what, err)
}
return exe, true, nil
}
@@ -153,11 +153,16 @@ func TestAComposeAskFailsLoudly(t *testing.T) {
}
// 1622, ADR 0615: the module asks the delivery indexer's read tool for the manifests a compose ask names by digest, and
// reads none of the indexer's state.
// reads none of the indexer's state. 1921 (B0): a run's input is asked of the checker seat's verb input, as the manifest
// invokes it, and no state of the checker's inputs is read (ADR 0596 §2).
func TestTheModuleReadsTheIndexersManifests(t *testing.T) {
if m := readManifest(t); !slices.Contains(m.Invokes, ManifestsTool) || slices.Contains(m.Reads, indexerModule+".manifests") {
m := readManifest(t)
if !slices.Contains(m.Invokes, ManifestsTool) || slices.Contains(m.Reads, indexerModule+".manifests") {
t.Errorf("invokes = %v, reads = %v: it asks %s", m.Invokes, m.Reads, ManifestsTool)
}
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 0615 §6: an indexer that does not answer is a finding of the module's own, and its next answer clears it.
@@ -391,6 +391,7 @@ func (m *Module) goEnvFor(dir string) []string {
// net for a caller that never released one (issue 1839, ADR 0509).
func (m *Module) CleanWorkspaces(now time.Time) {
m.sweepSaid(m.spaces().Sweep(now, buildstep.UnheldBound))
m.SweepKept(now, KeptBound)
}
// clock is the time now: Now when a test set it, else the wall clock.
@@ -0,0 +1,130 @@
package main
// **Kept programs have tools and a sweep** (novox/hq issue 1921, the seams review's must 2): the programs this module builds
// from a commit and keeps under its cache so each commit is built once on a node, the composers of compose asks and the
// programs of run asks, are read by build_go_kept, removed one by one by build_go_kept_drop, and swept when unused for
// KeptBound. A use touches the program's file, so its modification time is its last use.
import (
"fmt"
"os"
"path/filepath"
"sort"
"sync"
"time"
stdio "git.novox.be/novox/mesh-sdk/go"
)
// The tools of the kept programs.
const (
KeptTool = "build_go_kept"
KeptDropTool = "build_go_kept_drop"
)
// KeptBound is how long a kept program that no ask used is kept: a month, longer than any group lives, so only the build
// of a commit nobody asks about any more goes.
const KeptBound = 30 * 24 * time.Hour
// keptKinds are the cache directories of kept programs and the name of the program in each key's directory.
var keptKinds = map[string]string{"composers": "mesh-compose", "programs": "program"}
// KeptProgram is one kept program: its kind (the cache directory), its key, its size and when an ask last used it.
type KeptProgram struct {
Kind string `json:"kind"`
Key string `json:"key"`
Bytes int64 `json:"bytes"`
LastUsed time.Time `json:"last-used"`
}
// programPath is where the program of kind and key is kept.
func (m *Module) programPath(kind, key string) string {
return filepath.Join(m.Cache, kind, key, keptKinds[kind])
}
// lockOf is the mutex that guards the program at path against a build or a use while it is dropped.
func lockOf(path string) *sync.Mutex {
l, _ := composerLocks.LoadOrStore(path, &sync.Mutex{})
return l.(*sync.Mutex)
}
// touch records a use of the kept program at path.
func touch(path string, at time.Time) { _ = os.Chtimes(path, at, at) }
// Kept lists the kept programs, composers then programs, each kind by key.
func (m *Module) Kept() []KeptProgram {
out := []KeptProgram{}
for _, kind := range []string{"composers", "programs"} {
entries, _ := os.ReadDir(filepath.Join(m.Cache, kind))
for _, e := range entries {
if !e.IsDir() {
continue
}
st, err := os.Stat(m.programPath(kind, e.Name()))
if err != nil || !st.Mode().IsRegular() {
continue
}
out = append(out, KeptProgram{Kind: kind, Key: e.Name(), Bytes: st.Size(), LastUsed: st.ModTime().UTC()})
}
}
return out
}
// DropKept removes the kept program of kind and key; one already gone is no error.
func (m *Module) DropKept(kind, key string) error {
if _, ok := keptKinds[kind]; !ok {
return fmt.Errorf("a kept program is of kind composers or programs, not %q", kind)
}
if key == "" || key != filepath.Base(key) || key == "." || key == ".." {
return fmt.Errorf("%q is no kept program's key", key)
}
exe := m.programPath(kind, key)
l := lockOf(exe)
l.Lock()
defer l.Unlock()
return os.RemoveAll(filepath.Dir(exe))
}
// SweepKept removes the kept programs no ask used within bound before now, and answers what it removed.
func (m *Module) SweepKept(now time.Time, bound time.Duration) []KeptProgram {
var gone []KeptProgram
for _, k := range m.Kept() {
if now.Sub(k.LastUsed) <= bound {
continue
}
if m.DropKept(k.Kind, k.Key) == nil {
gone = append(gone, k)
}
}
sort.Slice(gone, func(i, j int) bool { return gone[i].Key < gone[j].Key })
if len(gone) > 0 && m.Say != nil {
m.Say("removed %d kept programs no ask used within %s", len(gone), bound)
}
return gone
}
// keptTools are the tools that read and drop the kept programs: no guard and no floor, since they only free room.
func (m *Module) keptTools() []stdio.Tool {
return []stdio.Tool{
{Name: KeptTool,
Description: "The programs this node keeps, built from a commit so each commit is built once: the composers of compose asks and " +
"the checking programs of run asks, each with its kind, key, size in bytes and when an ask last used it. One no ask used " +
"for 30 days is removed. Read-only.",
Run: func(map[string]any) (any, error) { return map[string]any{"kept": m.Kept()}, nil }},
{Name: KeptDropTool,
Description: "Remove one kept program by its kind (composers or programs) and key, as build_go_kept names them: the next ask " +
"for that commit builds it again. One already gone is no error.",
Input: map[string]any{"type": "object", "properties": map[string]any{
"kind": map[string]any{"type": "string", "enum": []string{"composers", "programs"}},
"key": map[string]any{"type": "string", "description": "the key build_go_kept names"},
}, "required": []string{"kind", "key"}},
Run: func(args map[string]any) (any, error) {
kind, _ := args["kind"].(string)
key, _ := args["key"].(string)
if err := m.DropKept(kind, key); err != nil {
return nil, err
}
return map[string]any{"dropped": key, "kind": kind}, nil
}},
}
}
@@ -0,0 +1,70 @@
package main
import (
"os"
"path/filepath"
"testing"
"time"
)
func keepProgram(t *testing.T, m *Module, kind, key string, size int, used time.Time) {
t.Helper()
exe := m.programPath(kind, key)
if err := os.MkdirAll(filepath.Dir(exe), 0o700); err != nil {
t.Fatal(err)
}
if err := os.WriteFile(exe, make([]byte, size), 0o700); err != nil {
t.Fatal(err)
}
touch(exe, used)
}
func toolOf(t *testing.T, m *Module, name string) func(map[string]any) (any, error) {
t.Helper()
for _, tool := range m.Tools(m.StepTools()) {
if tool.Name == name {
return tool.Run
}
}
t.Fatalf("no tool %s", name)
return nil
}
// Seams review must 2 (issue 1921): the kept composers and programs are listed with their size and last use, dropped one
// by one, and swept with the workspaces when unused past KeptBound; a use keeps a program.
func TestKeptProgramsAreListedDroppedAndSwept(t *testing.T) {
m, _ := fetchModule(t, &fakeStep{})
now := time.Now().UTC().Truncate(time.Second)
keepProgram(t, m, "composers", "old", 10, now.Add(-KeptBound-time.Hour))
keepProgram(t, m, "programs", "recent", 20, now.Add(-time.Hour))
keepProgram(t, m, "programs", "gone", 30, now)
out, err := toolOf(t, m, KeptTool)(nil)
if err != nil {
t.Fatal(err)
}
list := out.(map[string]any)["kept"].([]KeptProgram)
if len(list) != 3 || list[0].Kind != "composers" || list[0].Key != "old" || list[0].Bytes != 10 || !list[0].LastUsed.Equal(now.Add(-KeptBound-time.Hour)) {
t.Fatalf("the kept programs = %+v", list)
}
for _, bad := range []map[string]any{{"kind": "nope", "key": "gone"}, {"kind": "programs", "key": "../composers"}, {"kind": "programs", "key": ""}} {
if _, err := toolOf(t, m, KeptDropTool)(bad); err == nil {
t.Errorf("dropped with %v", bad)
}
}
for i := 0; i < 2; i++ { // a second drop is no error
if _, err := toolOf(t, m, KeptDropTool)(map[string]any{"kind": "programs", "key": "gone"}); err != nil {
t.Fatalf("drop %d: %v", i, err)
}
}
if _, err := os.Stat(m.programPath("programs", "gone")); !os.IsNotExist(err) {
t.Error("the dropped program is still kept")
}
// The sweep runs with the workspaces' (CleanWorkspaces, on its timer).
m.CleanWorkspaces(now)
if got := m.Kept(); len(got) != 1 || got[0].Key != "recent" {
t.Fatalf("kept after the sweep = %+v", got)
}
}
@@ -59,6 +59,7 @@ func main() {
m.Settings = m.Home + "/settings.json"
m.ReadKept = readOf(stdio.State(KeptState)).Get
m.ReadManifest = askManifest
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.
@@ -98,7 +98,12 @@ type Module struct {
// ReadManifest reads a key of the delivery indexer's manifests state (ManifestsState), which a compose ask's request
// names its manifests by (compose.go); GoModCache, when set, is the Go module cache in place of Cache's (a test).
ReadManifest func(key string) (json.RawMessage, error)
GoModCache string
// AskInput asks a seat verb the module invokes (module.json) as the runtime calls it (stdio.Ask): the verb a run ask's
// input names, answering the input by its digest (delivery.RunInput.Read).
AskInput func(key string, body any) (json.RawMessage, error)
// PlaceHead, when set, lays a run's head at a directory instead of fetching it (a test).
PlaceHead func(ctx context.Context, h delivery.RunHead, dir string) error
GoModCache string
spacesOnce sync.Once
workspaces *buildstep.Workspaces
@@ -115,7 +115,7 @@ func TestAGoThatCannotBeRunRefusesEveryStepAndHealthIsUnhealthy(t *testing.T) {
{Name: "short_probe", Run: run},
}, m.StepTools()...)
for _, tool := range m.Tools(steps) {
if tool.Name == StatusTool || tool.Name == HealthTool || tool.Name == StepReadTool || tool.Name == StepsReadTool || tool.Name == ToolchainsReadTool || tool.Name == ReleaseTool || tool.Name == StepEndTool {
if tool.Name == StatusTool || tool.Name == HealthTool || tool.Name == StepReadTool || tool.Name == StepsReadTool || tool.Name == ToolchainsReadTool || tool.Name == ReleaseTool || tool.Name == KeptTool || tool.Name == KeptDropTool || tool.Name == StepEndTool {
continue
}
var reason string
@@ -0,0 +1,132 @@
package main
// **A run ask runs a program built from a change, over heads it reads** (novox/hq ADR 0644, ADR 0656; issue 1921): the
// checking program of a group's check. The module builds the program at the commit the ask names, once per
// delivery.RunKey, as a compose ask builds a composer (programOf); fetches every head the ask names and lays each read-only
// under the directory MESH_RUN_HEADS names, at `<owner>/<name>`; asks the seat verb the ask's input names for it by its
// digest (as module.json invokes it; never a read of the asker's state, ADR 0596 §2) and writes it to the file
// MESH_RUN_INPUT names; and runs the program shut off with the ask's
// arguments: no network, none of the account. The answer is delivery.RunAnswer: the program's exit, the last line of its
// standard output and the first line of its standard error. A program that fails is an answer, not a fault of the step.
import (
"context"
"encoding/json"
"errors"
"fmt"
"os"
"path/filepath"
"strings"
"git.novox.be/novox/mesh-catalog/modules/mesh-build-go/buildstep"
sdkstep "git.novox.be/novox/mesh-sdk/go/buildstep"
"git.novox.be/novox/mesh-sdk/go/delivery"
)
// run runs a run ask: the program of its commit, built once, over its heads and its input.
func (m *Module) run(ctx context.Context, fetch buildstep.StepFunc, a delivery.StepAsk) (any, error) {
job := *a.Run
body, err := m.inputOf(job.Input)
if err != nil {
return nil, err
}
installed, err := m.Probe(ctx)
if err != nil {
return nil, err
}
key := delivery.RunKey(job.Program, delivery.ToolchainUsed{Version: installed, Recipe: m.Recipe.Version()}, a.Packages)
fetchArgs := sdkstep.FetchRequest{Repository: job.Program.Repository, Commit: job.Program.Commit, Packages: a.Packages}.Args()
exe, built, err := m.programOf(ctx, fetch, "programs", "checking program", key, fetchArgs, job.Program.Dir, job.Program.Package)
if err != nil {
return nil, err
}
ans, err := m.runProgram(ctx, exe, job, body)
if err != nil {
return nil, err
}
ans.Built, ans.Key = built, key
// The answer is written on the step key as one value: one over the bus's value limit (the input's bound) fails the
// step, naming its size, before it is written.
if raw, err := json.Marshal(ans); err != nil {
return nil, err
} else if len(raw) > delivery.RunInputBound {
return nil, fmt.Errorf("the checking program's answer is %d bytes, over the bound of %d bytes a step key holds", len(raw), delivery.RunInputBound)
}
return ans, nil
}
// inputOf is the input a run names, asked of the seat verb the ask names (module.json invokes it) and checked against
// its digest and size: bytes that are not the input named refuse the ask.
func (m *Module) inputOf(in delivery.RunInput) ([]byte, error) {
if m.AskInput == nil {
return nil, errors.New("this module asks no seat verb for a run's input")
}
raw, err := in.Read(m.AskInput)
if err != nil && raw != nil {
return nil, buildstep.Refuse("%v", err)
}
return raw, err
}
// placeHead lays one head's commit as a plain directory at dir, with no symlink in it; a test sets its own.
func (m *Module) placeHead(ctx context.Context, h delivery.RunHead, dir string) error {
if m.PlaceHead != nil {
return m.PlaceHead(ctx, h, dir)
}
if err := os.MkdirAll(dir, 0o700); err != nil {
return err
}
if err := buildstep.FetchCommit(ctx, m.Run, dir, m.remote(h.Repository), h.Commit, "", m.fetchOptions(ctx, h.Repository, h.Commit, "")...); err != nil {
return err
}
if err := m.dissociate(ctx, dir); err != nil {
return err
}
if link := firstSymlink(dir); link != "" {
return fmt.Errorf("%s holds a symlink, %s", h.Repository, link)
}
return nil
}
// runProgram lays the heads and the input in a workspace of its own and runs exe shut off over them.
func (m *Module) runProgram(ctx context.Context, exe string, job delivery.RunJob, input []byte) (delivery.RunAnswer, error) {
id := sdkstep.NewID()
done := m.spaces().Hold(id)
defer m.releaseSaid(id)
defer done()
ws := filepath.Join(m.Work, id)
heads := filepath.Join(ws, "heads")
for _, h := range job.Heads {
if err := m.placeHead(ctx, h, filepath.Join(heads, filepath.FromSlash(h.Repository))); err != nil {
return delivery.RunAnswer{}, fmt.Errorf("the head of %s at %s: %w", h.Repository, h.Commit, err)
}
}
if err := os.MkdirAll(heads, 0o700); err != nil {
return delivery.RunAnswer{}, err
}
file := filepath.Join(ws, "input")
if err := os.WriteFile(file, input, 0o600); err != nil {
return delivery.RunAnswer{}, err
}
run := m.Run
if !m.Inside {
iso := m.isolation(id, false)
iso.ReadOnly = append(iso.ReadOnly, exe)
iso.ReadOnlyInside = []string{heads} // bound after the workspace, so the program cannot change a head
run = buildstep.Isolated(m.Run, iso)
}
r := run(ctx, []string{delivery.RunHeadsEnv + "=" + heads, delivery.RunInputEnv + "=" + file}, exe, job.Args...)
if r.Err != "" && r.Status == 0 {
return delivery.RunAnswer{}, fmt.Errorf("the checking program could not run: %s", r.Err)
}
ans := delivery.RunAnswer{Exit: r.Status, Said: firstLine(r.Err + "\n" + r.Stderr)}
line := strings.TrimSpace(lastLine(r.Stdout))
switch {
case line == "":
case json.Valid([]byte(line)):
ans.Output = json.RawMessage(line)
default:
ans.Output, _ = json.Marshal(line)
}
return ans, nil
}
@@ -0,0 +1,141 @@
//go:build integration
package main
import (
"context"
"encoding/json"
"errors"
"net"
"os"
"path/filepath"
"strings"
"testing"
"time"
"github.com/nats-io/nats.go"
"github.com/nats-io/nats.go/jetstream"
"git.novox.be/novox/mesh-catalog/modules/mesh-build-go/buildstep"
"git.novox.be/novox/mesh-sdk/go/delivery"
"git.novox.be/novox/mesh-sdk/go/testbus"
)
// 1921/C3: a run ask put on the Go build seat's check stream is pulled by the module started whole on the real bus and
// answered on its step key <node>.<ask id>: the program, run shut off over its heads and its input (asked by digest of the
// seat verb the ask names, which a fake checker answers on the bus as the checker's verb input does), answers its last
// line and exit; a program cannot write under its heads, and one
// that dials the machine's network is shut off, and the answer says it failed.
func TestARunAskIsAnsweredOnTheBusAndItsProgramIsShutOff(t *testing.T) {
needTools(t, "the shut-off run", "bash", "bwrap")
nc, err := nats.Connect(testbus.URL(t))
if err != nil {
t.Fatal(err)
}
defer nc.Close()
js, err := jetstream.New(nc)
if err != nil {
t.Fatal(err)
}
ctx, stop := context.WithCancel(context.Background())
defer stop()
if err := buildstep.DeclareAskWorkers(ctx, js, delivery.AskGo, 2*time.Minute); err != nil {
t.Fatal(err)
}
t.Cleanup(func() { _ = js.DeleteStream(context.Background(), "SEAT_NODE_BUILD_GO") })
tcp, err := net.Listen("tcp", "127.0.0.1:0")
if err != nil {
t.Fatal(err)
}
defer tcp.Close()
states := newStates()
m := moduleWith(states, "go1.26.0")
m.Work, m.Cache, m.Home = t.TempDir(), t.TempDir(), t.TempDir()
m.Capacity = func() int { return 1 }
m.Run = StepRunner(m.Home)
for _, c := range m.caches() {
if err := os.MkdirAll(c, 0o700); err != nil {
t.Fatal(err)
}
}
m.PlaceHead = func(_ context.Context, h delivery.RunHead, dir string) error { return os.MkdirAll(dir, 0o700) }
body := `{"v":1}`
// The checker's verb input on the bus, and the module asking it as the runtime calls a seat verb it invokes.
asked := make(chan string, 4)
sub, err := nc.Subscribe("mesh.seat.mesh-delivery-checker.tool.input", func(msg *nats.Msg) {
var ask delivery.RunInputAsk
_ = json.Unmarshal(msg.Data, &ask)
asked <- ask.Digest
reply, _ := json.Marshal(map[string]any{"result": delivery.RunInputAnswer{Digest: ask.Digest, Input: []byte(body)}})
_ = msg.Respond(reply)
})
if err != nil {
t.Fatal(err)
}
defer sub.Unsubscribe()
m.AskInput = func(key string, b any) (json.RawMessage, error) {
if key != runInputVerb {
return nil, errors.New("asked " + key)
}
data, _ := json.Marshal(b)
msg, err := nc.Request("mesh.seat.mesh-delivery-checker.tool.input", data, 5*time.Second)
if err != nil {
return nil, err
}
var r struct {
Result json.RawMessage `json:"result"`
}
return r.Result, json.Unmarshal(msg.Data, &r)
}
a := runAskOf(t, body)
key := delivery.RunKey(a.Run.Program, delivery.ToolchainUsed{Version: "go1.26.0", Recipe: m.Recipe.Version()}, nil)
exe := m.programPath("programs", key)
os.MkdirAll(filepath.Dir(exe), 0o700)
script := "#!/bin/bash\n" + `if touch "$MESH_RUN_HEADS/novox/mesh-sdk/x" 2>/dev/null; then echo '{"wrote":true}'; exit 9; fi; echo "start"; exec 3<>/dev/tcp/127.0.0.1/` + strings.Split(tcp.Addr().String(), ":")[1] + " || exit 7\n" + `echo '{"dialed":true}'` + "\n"
if err := os.WriteFile(exe, []byte(script), 0o700); err != nil {
t.Fatal(err)
}
p := m.Puller(fakeSteps(new([]string)), &buildstep.JetStreamAsks{JS: js, Language: delivery.AskGo})
p.PullWait = 200 * time.Millisecond
go p.Pull(ctx)
raw, err := delivery.EncodeStepAsk(a)
if err != nil {
t.Fatal(err)
}
subject, _ := delivery.AskSubject(delivery.AskGo, delivery.AskCheck)
if _, err := js.Publish(ctx, subject, raw); err != nil {
t.Fatal(err)
}
stepKey, _ := delivery.AskStepKey("ace", a.ID)
var step delivery.Step
for deadline := time.Now().Add(60 * time.Second); time.Now().Before(deadline); time.Sleep(50 * time.Millisecond) {
states.mu.Lock()
v := states.steps[stepKey]
states.mu.Unlock()
if len(v) > 0 && v[len(v)-1].State != delivery.StepRunning {
step = v[len(v)-1]
break
}
}
if step.State != delivery.StepPassed {
t.Fatalf("the run ask ended %q: %s", step.State, step.Reason)
}
var ans delivery.RunAnswer
if err := json.Unmarshal(step.Result, &ans); err != nil {
t.Fatal(err)
}
if ans.Key != key || ans.Exit != 7 || strings.Contains(string(ans.Output), "dialed") || string(ans.Output) != `"start"` {
t.Fatalf("the answer of a program that dials out = %+v output %s", ans, ans.Output)
}
select {
case d := <-asked:
if d != a.Run.Input.Digest {
t.Errorf("the verb input was asked %s, want %s", d, a.Run.Input.Digest)
}
default:
t.Error("the run's input was not asked of the checker's verb input")
}
}
@@ -0,0 +1,154 @@
package main
import (
"context"
"crypto/sha256"
"encoding/hex"
"encoding/json"
"errors"
"fmt"
"os"
"path/filepath"
"strings"
"testing"
"time"
"git.novox.be/novox/mesh-catalog/modules/mesh-build-go/buildstep"
sdkstep "git.novox.be/novox/mesh-sdk/go/buildstep"
"git.novox.be/novox/mesh-sdk/go/composer"
"git.novox.be/novox/mesh-sdk/go/delivery"
"git.novox.be/novox/mesh-sdk/go/forgekey"
)
const runCommit = "5555555555555555555555555555555555555555"
// runInputVerb is the seat verb a run ask's input is served on, as the checker names it in the ask.
const runInputVerb = "seat:mesh-delivery-checker.input"
// answersInput is a seat verb serving body by its digest, as the checker's verb input answers (delivery.RunInputAnswer).
func answersInput(body string) func(key string, b any) (json.RawMessage, error) {
return func(key string, b any) (json.RawMessage, error) {
ask, _ := b.(delivery.RunInputAsk)
if sum := sha256.Sum256([]byte(body)); key != runInputVerb || ask.Digest != hex.EncodeToString(sum[:]) {
return nil, fmt.Errorf("the checker keeps no input %s", ask.Digest)
}
return json.Marshal(delivery.RunInputAnswer{Digest: ask.Digest, Input: []byte(body)})
}
}
func runAskOf(t *testing.T, body string) delivery.StepAsk {
t.Helper()
in, err := delivery.RunInputOf(runInputVerb, []byte(body))
if err != nil {
t.Fatal(err)
}
return delivery.StepAsk{ID: sdkstep.NewID(), Kind: delivery.AskCheck, Language: delivery.AskGo, Asker: "mesh-delivery-checker", At: time.Now(),
Run: &delivery.RunJob{Program: delivery.RunProgram{Repository: "novox/mesh-delivery", Commit: runCommit, Dir: ".", Package: "./cmd/prog"},
Heads: []delivery.RunHead{{Repository: "novox/mesh-sdk", Commit: runCommit}, {Repository: "novox/mesh-host", Commit: runCommit}},
Input: in, Args: []string{"-seats"}}}
}
// 1921/C1: a kept checking program is run over the input checked against its digest and the heads laid under
// MESH_RUN_HEADS at <owner>/<name>, and the answer carries its exit and last line; nothing is fetched or built; a
// program that fails, or says no JSON, is an answer; an input that is not the one named is refused; and the program was
// kept under delivery.RunKey, so a second ask finds it.
func TestAKeptProgramIsRunOverItsHeadsAndInput(t *testing.T) {
m := composeModule(t)
body := `{"v":1,"case":"x"}`
m.AskInput = answersInput(body)
m.PlaceHead = func(_ context.Context, h delivery.RunHead, dir string) error {
return os.MkdirAll(dir, 0o700)
}
a := runAskOf(t, body)
key := delivery.RunKey(a.Run.Program, delivery.ToolchainUsed{Version: "go1.26.0", Recipe: m.Recipe.Version()}, nil)
exe := m.programPath("programs", key)
os.MkdirAll(filepath.Dir(exe), 0o700)
script := "#!/bin/sh\n" +
`echo "noise"; echo "args $*" >&2; ` +
`printf '{"input":%s,"heads":"%s"}\n' "$(cat "$MESH_RUN_INPUT")" "$(cd "$MESH_RUN_HEADS" && ls -d */* | tr '\n' ' ')"; exit 3` + "\n"
if err := os.WriteFile(exe, []byte(script), 0o700); err != nil {
t.Fatal(err)
}
fetched := 0
fetch := func(context.Context, map[string]any) (any, error) { fetched++; return nil, nil }
out, err := m.run(context.Background(), fetch, a)
if err != nil {
t.Fatal(err)
}
ans := out.(delivery.RunAnswer)
if ans.Built || ans.Key != key || fetched != 0 || ans.Exit != 3 || ans.Said != "args -seats" ||
string(ans.Output) != `{"input":{"v":1,"case":"x"},"heads":"novox/mesh-host novox/mesh-sdk "}` {
t.Fatalf("answer = %+v (fetched %d) output %s", ans, fetched, ans.Output)
}
if entries, _ := os.ReadDir(m.Work); len(entries) != 0 {
t.Errorf("the run's workspace was left: %v", entries)
}
// Shut off, the heads are bound read-only after the writable workspace, so a write under MESH_RUN_HEADS fails: a
// read-only bind before the workspace's is covered by it.
m.Inside = false
var argv []string
m.Run = func(_ context.Context, _ []string, _ string, args ...string) forgekey.Ran {
argv = args
return forgekey.Ran{Status: 0, Stdout: "{}"}
}
if _, err := m.run(context.Background(), fetch, a); err != nil {
t.Fatal(err)
}
line, ws, heads := " "+strings.Join(argv, " ")+" ", -1, -1
for i := 0; i+2 < len(argv); i++ {
switch {
case argv[i] == "--bind" && filepath.Dir(argv[i+1]) == m.Work:
ws = i
case argv[i] == "--ro-bind" && filepath.Base(argv[i+1]) == "heads" && argv[i+2] == argv[i+1]:
heads = i
}
}
if ws < 0 || heads < ws || !strings.Contains(line, " "+delivery.RunHeadsEnv+" "+argv[heads+1]+" ") {
t.Errorf("the heads are not bound read-only over the workspace: %s", line)
}
// An answer over the bus's value limit fails the step, naming its size.
m.Run = func(context.Context, []string, string, ...string) forgekey.Ran {
return forgekey.Ran{Status: 0, Stdout: `"` + strings.Repeat("x", delivery.RunInputBound) + `"`}
}
if _, err := m.run(context.Background(), fetch, a); err == nil || !strings.Contains(err.Error(), "over the bound") {
t.Errorf("an answer over the bound: %v", err)
}
m.Inside, m.Run = true, StepRunner(t.TempDir())
// An input other than the one named is refused, naming it; one the verb does not answer fails the step, saying why.
m.AskInput = func(_ string, b any) (json.RawMessage, error) {
return json.Marshal(delivery.RunInputAnswer{Digest: b.(delivery.RunInputAsk).Digest, Input: []byte(`{"other":1}`)})
}
if _, err := m.run(context.Background(), fetch, a); err == nil || !strings.Contains(err.Error(), "not the one the ask names") || !errors.As(err, new(*buildstep.Refusal)) {
t.Errorf("a wrong input: %v", err)
}
m.AskInput = answersInput(`{"v":2}`)
if _, err := m.run(context.Background(), fetch, a); err == nil || !strings.Contains(err.Error(), "keeps no input") {
t.Errorf("an input the verb does not answer: %v", err)
}
}
// 1921/C2 (F2): a compose ask's seats reach the composer: the module decodes and fills the request and writes it again,
// and used to drop the seats.
func TestAComposeAskKeepsItsSeats(t *testing.T) {
m := composeModule(t)
read, digests := indexerHolds(t, `{"module":"demo","version":"1"}`)
m.ReadManifest = read
a := composeAskOf(t, digests[0])
req := composer.Request{V: composer.Version, Seats: []json.RawMessage{json.RawMessage(`{"seat":"node-x"}`)},
Cases: []composer.Case{{Name: "change", Manifests: map[string]composer.Manifest{"demo": {Digest: digests[0]}}}}}
a.Compose.Request, _ = json.Marshal(req)
key := delivery.ComposerKey("novox/mesh-sdk", composeCommit, "go", composer.Package, delivery.ToolchainUsed{Version: "go1.26.0", Recipe: m.Recipe.Version()}, nil)
exe := m.composerPath(key)
os.MkdirAll(filepath.Dir(exe), 0o700)
script := "#!/bin/sh\n" + `if grep -q '"seats":\[{"seat":"node-x"}\]' "$2"; then S=true; else S=false; fi; ` +
`echo '{"v":1,"sdk":"stand-in","cases":[{"name":"change","manifests":{"demo":{"accepted":'$S'}}}]}'` + "\n"
os.WriteFile(exe, []byte(script), 0o700)
out, err := m.compose(context.Background(), nil, a)
if err != nil {
t.Fatal(err)
}
var v composer.Verdict
if err := json.Unmarshal(out.(delivery.ComposeAnswer).Verdict, &v); err != nil || !v.Cases[0].Manifests["demo"].Accepted {
t.Fatalf("the composer did not receive the seats: %+v %v", v, err)
}
}
@@ -45,6 +45,7 @@ func (m *Module) Tools(steps []StepTool) []stdio.Tool {
tools = append(tools, m.ReadTools()...)
tools = append(tools, buildstep.StepEndTool(StepEndTool, m.Node, m.stepScan(), m.Held))
tools = append(tools, m.releaseTool())
tools = append(tools, m.keptTools()...)
return append(tools,
stdio.Tool{Name: StatusTool,
Description: "What the Go build module is on this machine: the Go version its recipe builds with and the one installed, " +
+4 -1
View File
@@ -34,7 +34,8 @@
"seat:oci-artifact-registry.publish",
"seat:oci-artifact-registry.publish-part",
"seat:go-package-registry.publish",
"seat:go-package-registry.publish-part"
"seat:go-package-registry.publish-part",
"seat:mesh-delivery-checker.input"
],
"config": {
"capacity": {
@@ -63,6 +64,8 @@
"build_go_build",
"build_go_imports",
"build_go_release",
"build_go_kept",
"build_go_kept_drop",
"build_go_step",
"build_go_steps",
"build_go_step_end",
+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 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 pending version's reviews with their findings (`pending-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,6 +1,8 @@
package main
import (
"context"
"fmt"
"reflect"
"strings"
"testing"
@@ -189,6 +191,33 @@ 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,14 +312,17 @@ 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"`
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"`
// 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"`
// Brief is what the operator's ask shows of it (issue 639).
Brief *Brief `json:"brief,omitempty"`
Cut string `json:"cut,omitempty"`
@@ -371,7 +374,17 @@ 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
@@ -452,6 +465,10 @@ 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
@@ -107,9 +107,12 @@ func newBusWorld(t *testing.T) *busWorld {
return map[string]any{"folders": f, "documents": d}, err
})
answer("records.records_read", func(body json.RawMessage) (any, error) {
var a struct{ Path string }
var a struct {
Path string
Head int
}
_ = json.Unmarshal(body, &a)
c, commit, err := records.Read(a.Path)
c, commit, err := records.ReadHead(a.Path, a.Head)
return map[string]any{"content": c, "commit": commit}, err
})
j := &judge{}
@@ -24,8 +24,26 @@ type Design struct {
Status string
Decisions []string
Milestones []Milestone
// Unread says why the design's frontmatter and title could not be read, when they could not (novox/hq issue
// 1947); the progress answer carries it as a warning beside a tree built from the tracker's milestone records.
Unread string
}
// unreadWarnings says each design whose head was not read, so none drops out of an answer silently.
func unreadWarnings(ds []*Design) []string {
var out []string
for _, d := range ds {
if d.Unread != "" {
out = append(out, d.Unread)
}
}
return out
}
// DesignHeadBytes is how much of a design is read: its frontmatter and its title lead it, and its milestones are the
// tracker's records, so the rest is never needed.
const DesignHeadBytes = 16 << 10
// Milestone is one milestone of a design, as the issue tracker keeps it.
type Milestone struct {
ID string
@@ -51,12 +69,13 @@ func parseDesign(path, content string) *Design {
d.Number, _ = strconv.Atoi(m[1])
}
lines := strings.Split(content, "\n")
body := 0
body, closed := 0, false
if len(lines) > 0 && strings.TrimSpace(lines[0]) == "---" {
for i := 1; i < len(lines); i++ {
l := lines[i]
if strings.TrimSpace(l) == "---" {
body = i + 1
closed = true
break
}
if v, ok := strings.CutPrefix(l, "status:"); ok {
@@ -67,6 +86,9 @@ func parseDesign(path, content string) *Design {
}
}
}
if len(lines) > 0 && strings.TrimSpace(lines[0]) == "---" && !closed {
d.Unread = fmt.Sprintf("the frontmatter of %s does not close in the part read: its state and decisions are not known here", path)
}
for i := body; i < len(lines) && d.Title == ""; i++ {
if m := titleLine.FindStringSubmatch(strings.TrimSpace(lines[i])); m != nil {
d.Title = strings.TrimSpace(m[1])
@@ -151,12 +173,22 @@ func (k *KB) designs(all bool, number int) ([]*Design, string, error) {
}
d := c.byPath[p]
if d == nil {
content, _, err := k.Records.Read(p)
var content string
var err error
if h, ok := k.Records.(HeadReader); ok {
content, _, err = h.ReadHead(p, DesignHeadBytes)
} else {
content, _, err = k.Records.Read(p)
}
if err != nil {
return nil, "", fmt.Errorf("the records module does not read %s: %v", p, err)
d = parseDesign(p, "")
d.Unread = fmt.Sprintf("the records module does not read %s (%v): its title and state are not known here", p, err)
} else {
d = parseDesign(p, content)
}
if d.Unread == "" {
c.byPath[p] = d
}
d = parseDesign(p, content)
c.byPath[p] = d
k.progReads++
}
out = append(out, d)
@@ -22,6 +22,12 @@ type Records interface {
Read(path string) (content, commit string, err error)
}
// HeadReader is Records that can read only the first part of a document: a caller that needs a frontmatter or a
// title reads a long document through it, bounded, never whole (novox/hq issue 1947).
type HeadReader interface {
ReadHead(path string, bytes int) (content, commit string, err error)
}
// busRecords is Records over the bus: the records module's tools, called as this module.
type busRecords struct{}
@@ -58,8 +64,18 @@ func (busRecords) List(folder string) ([]string, []string, error) {
return f.Folders, f.Documents, nil
}
func (busRecords) Read(path string) (string, string, error) {
raw, err := stdio.Ask("records.records_read", map[string]any{"path": path})
func (busRecords) Read(path string) (string, string, error) { return busRead(path, 0) }
func (busRecords) ReadHead(path string, bytes int) (string, string, error) {
return busRead(path, bytes)
}
func busRead(path string, head int) (string, string, error) {
args := map[string]any{"path": path}
if head > 0 {
args["head"] = head
}
raw, err := stdio.Ask("records.records_read", args)
if err != nil {
return "", "", err
}
@@ -118,11 +134,19 @@ func (d dirRecords) List(folder string) ([]string, []string, error) {
return folders, docs, nil
}
func (d dirRecords) Read(path string) (string, string, error) {
func (d dirRecords) Read(path string) (string, string, error) { return d.ReadHead(path, 0) }
func (d dirRecords) ReadHead(path string, head int) (string, string, error) {
raw, err := os.ReadFile(filepath.Join(d.dir, filepath.FromSlash(path)))
if err != nil {
return "", "", err
}
if head > 0 && head < len(raw) {
raw = raw[:head]
if i := strings.LastIndexByte(string(raw), '\n'); i > 0 {
raw = raw[:i+1]
}
}
commit, _ := d.Standing()
return string(raw), commit, nil
}
@@ -841,7 +841,7 @@ func verbTools(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 operator's ask when it waits for one. " +
"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 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.",
@@ -329,6 +329,12 @@ func (k *KB) progressDesign(p *Progress, number int, milestone string) error {
return err
}
d := ds[0]
if d.Unread != "" {
p.Warnings = append(p.Warnings, d.Unread)
if d.Title == "" {
d.Title = fmt.Sprintf("design %d", d.Number)
}
}
if len(d.Milestones) == 0 {
return fmt.Errorf("design %d has no milestone records in the issue tracker: seed them from its table with "+
"issue-tracker.import milestones %d", number, number)
@@ -387,6 +393,7 @@ func (k *KB) progressMesh(p *Progress) error {
return err
}
p.Source = fmt.Sprintf("the issue tracker's milestone records, and the designs in %s at hq %s", DesignsFolder, short(commit))
p.Warnings = append(p.Warnings, unreadWarnings(ds)...)
var with []*Design
var without []string
for _, d := range ds {
@@ -514,6 +521,7 @@ func (k *KB) progressRecord(p *Progress, kind string, n int) error {
return err
}
p.Source = fmt.Sprintf("the knowledge base, the designs at hq %s and the issue tracker's links", short(commit))
p.Warnings = append(p.Warnings, unreadWarnings(ds)...)
var designs []string
for _, d := range ds {
for _, dn := range d.Decisions {
@@ -42,6 +42,15 @@ func (w withDesigns) Read(path string) (string, string, error) {
return c, commit, err
}
func (w withDesigns) ReadHead(path string, n int) (string, string, error) {
if c, commit, err := w.base.ReadHead(path, n); err == nil {
return c, commit, nil
}
c, _, err := w.designs.ReadHead(path, n)
commit, _ := w.base.Standing()
return c, commit, err
}
// seatsWorld plays the issue tracker and mesh-delivery from the captured answers, and counts what it is asked.
type seatsWorld struct {
mu sync.Mutex
@@ -142,6 +151,19 @@ func (c *countingRecords) Read(p string) (string, string, error) {
return content, c.commit, err
}
func (c *countingRecords) ReadHead(p string, n int) (string, string, error) {
c.reads++
content, _, err := c.dirRecords.ReadHead(p, n)
return content, c.commit, err
}
// brokenRead is Records whose reads fail as the records module's do for a document longer than it answers whole.
type brokenRead struct{ dirRecords }
func (brokenRead) ReadHead(p string, n int) (string, string, error) {
return "", "", fmt.Errorf("%s is longer than the records module answers whole", p)
}
// 1947/AC2, as issue 1781 changed it: a design's title, state and decisions are hq's; its milestones are the issue
// tracker's records, ordered as its table, an id found with spaces or hyphens, and a closed one holding no issue done.
func TestADesignsMilestonesAreTheTrackersRecords(t *testing.T) {
@@ -316,6 +338,18 @@ func TestProgressOfADesignMinimalAndDetailed(t *testing.T) {
if big, _ := k.Progress(ProgressAsk{Of: "design 47", Bound: 8192}); big.Cut != "" || big.Tree != p.Tree {
t.Errorf("a minimal answer at bound 8192 is cut: %q", big.Cut)
}
// A design whose head the records module cannot read is said in a warning beside the tracker-built tree, never a failure.
kb := &KB{Records: brokenRead{progressRecords()}, Seats: newSeatsWorld(t).ask, Now: k.Now}
if bp, err := kb.Progress(ProgressAsk{Of: "design 47"}); err != nil || !strings.Contains(strings.Join(bp.Warnings, "\n"), "does not read") ||
!strings.Contains(bp.Tree, "design 47") || !strings.Contains(bp.Tree, "R6") {
t.Fatalf("an unreadable design: %v %+v", err, bp)
}
if mp, err := kb.Progress(ProgressAsk{Of: "mesh"}); err != nil || !strings.Contains(strings.Join(mp.Warnings, "\n"), "does not read") {
t.Fatalf("the mesh level drops an unread design silently: %v %v", err, mp.Warnings)
}
if d := parseDesign(DesignsFolder+"/47-x.md", "---\nstatus: in-progress\n"); d.Unread == "" {
t.Errorf("a frontmatter cut before its close is not said")
}
lines := strings.Split(p.Tree, "\n")
if !strings.HasPrefix(lines[0], "✗ design 47 · Delivery: from a commit to delivered — ") {
t.Fatalf("first line %q", lines[0])
@@ -361,6 +361,15 @@ 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 {
+22 -4
View File
@@ -1480,27 +1480,35 @@ func (h *Holder) sayClosed(key string, rec AskRecord, outcome string, words stri
}
// 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.
// ended (all takes both), Since the asks opened at or after that time, Bound the most bytes of the answer's asks. With no
// argument at all the query is the open asks within DefaultAsksBound (issue 1974), and the answer says so.
type AsksQuery struct {
Key, State string
Since time.Time
Bound int
// Defaulted is set by AsksQueryOf when no key, state or since was given: the open asks only.
Defaulted bool
}
// DefaultAsksBound is the most bytes of asks an answer holds when the caller gives no bound (issue 1974: the whole of a
// month's asks was 737,934 characters, more than a session reads).
const DefaultAsksBound = 32768
// 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"`
// Default says that no key, state or since was given, so only the open asks are answered, and how to ask for more.
Default string `json:"default,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 q.State != "" && q.State != "open" && q.State != "ended" && q.State != "all" {
return q, errorf("state is open, ended or all, not %q", q.State)
}
if since := strArg(a, "since"); since != "" {
t, err := time.Parse(time.RFC3339, since)
@@ -1515,6 +1523,12 @@ func AsksQueryOf(a map[string]any) (AsksQuery, error) {
}
q.Bound = int(v)
}
if q.Key == "" && q.State == "" && q.Since.IsZero() {
q.State, q.Defaulted = "open", true
}
if q.Bound == 0 {
q.Bound = DefaultAsksBound
}
return q, nil
}
@@ -1523,6 +1537,10 @@ func AsksQueryOf(a map[string]any) (AsksQuery, error) {
func (q AsksQuery) Narrow(all []AskView) AsksAnswer {
var out AsksAnswer
out.Asks = []AskView{}
if q.Defaulted {
out.Default = "no key, state or since was given, so only the open asks are shown: state ended or all for the ended ones, " +
"key for one ask, since for the asks opened from a time"
}
var kept []AskView
for _, v := range all {
open := v.State == StateOpen || v.State == StateHeld
+8 -9
View File
@@ -143,7 +143,7 @@ func TestAHeldAskNamesEveryKindAndWhyWhenNoChannelIsReady(t *testing.T) {
}
// 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.
// bound, said when it cut the answer short; with no argument it answers the open asks only, within a default bound, and says so (issue 1974).
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}
@@ -159,13 +159,15 @@ func TestTheAsksVerbNarrowsItsAnswer(t *testing.T) {
cut bool
err bool
}{
{"no argument is the zero query", map[string]any{}, nil, 0, false, false},
{"no argument is the open asks, said", map[string]any{}, []string{"a.3", "a.2"}, 2, false, false},
{"state all is every ask", map[string]any{"state": "all"}, []string{"a.3", "a.2", "a.1", "a.0"}, 4, false, false},
{"a key finds an ended ask whatever its state", map[string]any{"key": "a.0"}, []string{"a.0"}, 1, 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 bound under the asks cuts", map[string]any{"bound": float64(len(one) + 5)}, []string{"a.3"}, 2, 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},
} {
@@ -177,13 +179,10 @@ func TestTheAsksVerbNarrowsItsAnswer(t *testing.T) {
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)
if (got.Default != "") != (tc.args["key"] == nil && tc.args["state"] == nil && tc.args["since"] == nil) || q.Bound == 0 {
t.Errorf("%s: default %q, bound %d: only no argument is defaulted, and every query is bounded", tc.name, got.Default, q.Bound)
}
var keys []string
for _, v := range got.Asks {
keys = append(keys, v.Key)
+6 -8
View File
@@ -735,12 +735,13 @@ 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.",
"ended (the warrant: who chose what, through which channel), and whether its asker was told. With no argument " +
"only the open asks, within 32768 bytes; the answer says what it cut and how to ask for more: key for one ask " +
"(any state), state open, ended or all, since (a time, RFC 3339) for the asks opened from then, and bound, the " +
"most bytes of the answer.",
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"),
"key": str("one ask, by its key (asker.id); none for the open asks"),
"state": str("open (open and held), ended or all; none is open when no key or since is given, else 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) {
@@ -748,9 +749,6 @@ func tools(h *Holder, l *listening) []stdio.Tool {
if err != nil {
return nil, err
}
if q == (AsksQuery{}) {
return h.AsksNow(), nil
}
return q.Narrow(h.AsksNow()), nil
}},
{Name: "messenger_identities",
+10 -3
View File
@@ -52,9 +52,16 @@ func tools(r *Records) []stdio.Tool {
return r.Search(strArg(a, "query"), limit)
}},
{Name: "records_read",
Description: "One document, whole, by its path in the repository — a decision record, a design document, an issue report.",
Input: map[string]any{"path": str("the document's path, e.g. 02-DECISIONS/0025-....md")},
Run: func(a map[string]any) (any, error) { return r.Read(strArg(a, "path")) }},
Description: "One document by its path (whole, or its first part with head) in the repository — a decision record, a design document, an issue report.",
Input: map[string]any{"path": str("the document's path, e.g. 02-DECISIONS/0025-....md"),
"head": map[string]any{"type": "number", "description": "only the first this many bytes, cut at a line end (optional); the answer says partial"}},
Run: func(a map[string]any) (any, error) {
head := 0
if v, ok := a["head"].(float64); ok {
head = int(v)
}
return r.ReadHead(strArg(a, "path"), head)
}},
{Name: "records_list",
Description: "What a folder of the repository holds: its sub-folders and its documents. The root when no folder is named.",
Input: map[string]any{"folder": str("a folder inside the repository (optional)")},
+17 -2
View File
@@ -322,12 +322,18 @@ type Document struct {
Path string `json:"path"`
Content string `json:"content"`
Truncated bool `json:"truncated"`
Commit string `json:"commit"`
// Partial is true when only the first part the caller asked for (head) is answered; Truncated stays the answer's own cap.
Partial bool `json:"partial,omitempty"`
Commit string `json:"commit"`
}
// Read is one document, whole, or its first part with a note when it is very long. The path stays inside
// the checkout: `..` and absolute paths are refused, not resolved.
func (r *Records) Read(p string) (Document, error) {
func (r *Records) Read(p string) (Document, error) { return r.ReadHead(p, 0) }
// ReadHead is Read, or with head above zero only the first head bytes of the document, cut at a line end, so a
// caller that needs a frontmatter or a title reads a long document without its whole length (novox/hq issue 1947).
func (r *Records) ReadHead(p string, head int) (Document, error) {
clean, ok := inside(p)
if !ok || clean == "" {
return Document{}, fmt.Errorf("%q is not a path inside the repository", p)
@@ -337,6 +343,15 @@ func (r *Records) Read(p string) (Document, error) {
return Document{}, fmt.Errorf("the repository holds no %s — `records_list` says what it holds", clean)
}
d := Document{Path: clean, Content: string(raw), Commit: r.Standing().Commit}
head = min(head, MostBytes)
if head > 0 && head < len(d.Content) {
cut := d.Content[:head]
if i := strings.LastIndex(cut, "\n"); i > 0 {
cut = cut[:i+1]
}
d.Content, d.Partial = cut, true
return d, nil
}
if len(d.Content) > MostBytes {
d.Content = d.Content[:MostBytes] + "\n\n[… truncated; the document is longer than this answer carries]"
d.Truncated = true
@@ -70,6 +70,10 @@ func TestAPhraseInOneDesignDocumentComesBackFromWhereItIsWritten(t *testing.T) {
if err != nil || !strings.Contains(doc.Content, "We decided the thing") || doc.Truncated {
t.Fatalf("%+v %v", doc, err)
}
if head, err := r.ReadHead("02-DECISIONS/0001-a-decision.md", 30); err != nil || !head.Partial || head.Truncated || len(head.Content) > 30 ||
!strings.HasSuffix(head.Content, "\n") || !strings.HasPrefix(doc.Content, head.Content) {
t.Fatalf("a head read: %+v %v", head, err)
}
root, _ := r.List("")
if strings.Join(root.Folders, ",") != "02-DECISIONS,03-DESIGN" || strings.Join(root.Documents, ",") != "README.md" {
t.Fatalf("%+v", root)