Compare commits

..
Author SHA1 Message Date
mesh-tools 980ade9338 Merge pull request 'Progress reads only a design's head, so a long design never takes the progress view down (issue 1947)' (#623) from fix/1947-progress-bounded-read into main 2026-11-06 23:41:28 +00:00
jschoubben 7c0ce2fe21 Progress: an unread design is not cached and is warned of at every level; head is capped (issue 1947)
mesh/check being carried to 7c0ce2fe218b on 980ade933853 | from 1da83c29ddf9
mesh/delivery merged: source.main moved to novox/mesh-catalog@980ade933853
2026-11-07 00:25:02 +01:00
jschoubben be7d649d1f Progress reads only a design's head, and says so beside the tracker's tree when even that fails (issue 1947) 2026-11-07 00:21:17 +01:00
mesh-tools 1da83c29dd Merge pull request 'operator-channel.asks answers open asks within a bound when asked with no argument (issue 1974)' (#621) from fix/1974-asks-default-bounded into main 2026-11-06 21:05:59 +00:00
jschoubben b6d85847f8 operator-channel.asks answers the open asks within a default bound, and says what it cut (issue 1974)
mesh/check being carried to b6d85847f855 on 1da83c29ddf9 | from 1d07778c88dc
mesh/delivery merged: source.main moved to novox/mesh-catalog@1da83c29ddf9
Why: with no argument the verb answered every ask of the month, 737,934 characters, more than a session reads.

Claude-Session: https://claude.ai/code/session_01NcbgfFSnn24M3GcqttoU7i
2026-11-06 21:51:44 +01:00
29 changed files with 207 additions and 799 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.
@@ -98,25 +98,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 +125,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 +137,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
}
@@ -150,17 +150,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
@@ -348,7 +348,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,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, " +
-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",
@@ -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
}
@@ -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])
+22 -4
View File
@@ -1418,27 +1418,35 @@ func (h *Holder) AsksNow() []AskView {
}
// 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)
@@ -1453,6 +1461,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
}
@@ -1461,6 +1475,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
@@ -734,12 +734,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) {
@@ -747,9 +748,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)