Pull the Go build seat's asks beside the tools, delivery first and at most the node's capacity, so no part chooses a node for a Go build or check (issue 1887 A3go, ADR 0525 §2-§6)

modules/mesh-build-go/cmd/mesh-build-go/asks.go

@@ -0,0 +1,133 @@
+package main
+
+import (
+	"context"
+	"encoding/json"
+	"fmt"
+
+	stdio "git.novox.be/novox/mesh-sdk/go"
+	"git.novox.be/novox/mesh-sdk/go/buildstep"
+	"git.novox.be/novox/mesh-sdk/go/delivery"
+)
+
+// **The module pulls its seat's asks** (novox/hq ADR 0525 §2-§6, ADR 0536 §4; issue 1887 A3go). It holds the seat
+// node-build-go on its node and pulls the seat's asks through its runtime (buildstep.SeatAsks): delivery first, then
+// check, then recheck, at most the node's build capacity at once (its setting `capacity`, read at every pull). Each ask
+// runs with the same step bodies its tools run, behind the same guard, and ends on the key <node>.<ask id> of its steps
+// state. The tools stay as the hand path (ADR 0525 §8).
+
+// Seat is the node seat this module holds.
+const Seat = "node-build-go"
+
+// KeptState is the checker's kept verdicts, which a check ask takes for the packages whose digest has one (issue
+// 1887's design limit 1).
+const KeptState = delivery.CheckerModule + ".kept"
+
+// askSteps are the module's step bodies for an ask, each holding its workspace while it runs, as a tool's does.
+func (m *Module) askSteps(steps []StepTool) buildstep.AskSteps {
+	byName := map[string]buildstep.StepFunc{}
+	for _, s := range steps {
+		run := s.Run
+		byName[s.Name] = func(ctx context.Context, args map[string]any) (any, error) {
+			defer m.holdArg(args)()
+			return run(ctx, args)
+		}
+	}
+	return buildstep.AskSteps{
+		Fetch: byName["build_go_fetch"], Build: byName["build_go_build"], Digest: byName["build_go_digest"], Test: byName["build_go_test"],
+		Toolchain: func(ctx context.Context) (delivery.ToolchainUsed, error) {
+			installed, err := m.Probe(ctx)
+			if err != nil {
+				return delivery.ToolchainUsed{}, err
+			}
+			return delivery.ToolchainUsed{Version: installed, Recipe: m.Recipe.Version()}, nil
+		},
+		Kept:    m.KeptLookup,
+		Release: func(ws string) error { return m.spaces().Release(ws) },
+		Logf:    m.Say,
+	}
+}
+
+// keptValue reads one key of the checker's kept state; nil when it is absent. Main sets it to the state's read; a test
+// its own.
+func (m *Module) keptValue(key string) (json.RawMessage, error) {
+	if m.ReadKept == nil {
+		return nil, nil
+	}
+	return m.ReadKept(key)
+}
+
+// KeptLookup is whether the checker kept a verdict for the level and digest, and how many tests it counted.
+func (m *Module) KeptLookup(ctx context.Context, level, digest string) (int, bool, error) {
+	key, err := delivery.KeptKey(level, digest)
+	if err != nil {
+		return 0, false, err
+	}
+	raw, err := m.keptValue(key)
+	if err != nil || raw == nil {
+		return 0, false, err
+	}
+	v, err := delivery.DecodeKept(raw)
+	if err != nil {
+		return 0, false, fmt.Errorf("%s holds what is not a kept verdict: %w", key, err)
+	}
+	return v.Tests, true, nil
+}
+
+// Puller is the module's pull of its seat's asks, on source (the runtime's seat pull when nil).
+func (m *Module) Puller(steps []StepTool, source buildstep.AskSource) *buildstep.Puller {
+	if source == nil {
+		source = buildstep.SeatAsks{Language: delivery.AskGo}
+	}
+	capacity := m.Capacity
+	if capacity == nil {
+		capacity = buildstep.NodeCapacity(m.Settings, buildstep.DefaultCapacities, m.Node, m.Say)
+	}
+	ask := m.askSteps(steps)
+	var load func(string) ([]byte, error)
+	if m.ReadSteps.Get != nil {
+		load = func(key string) ([]byte, error) {
+			raw, err := m.ReadSteps.Get(key)
+			return raw, err
+		}
+	}
+	return &buildstep.Puller{Source: source, Language: delivery.AskGo, Node: m.Node, Put: m.Steps, Load: load,
+		Capacity: capacity, Logf: m.Say,
+		Run: func(ctx context.Context, a delivery.StepAsk) (any, error) {
+			if err := m.Check(ctx); err != nil {
+				return nil, err
+			}
+			if err := m.room(); err != nil {
+				return nil, err
+			}
+			return buildstep.RunAsk(ctx, ask, a)
+		}}
+}
+
+// SeatTools are the seat's verbs on this node (ADR 0525 §10): what it runs, and its pause and resume.
+func (m *Module) SeatTools(p *buildstep.Puller) []stdio.Tool {
+	verb := func(name, description string, run func() (any, error)) stdio.Tool {
+		return stdio.Tool{Name: Seat + "." + name, Description: description,
+			Input: map[string]any{"type": "object", "properties": map[string]any{}},
+			Run:   func(map[string]any) (any, error) { return run() }}
+	}
+	current := func() (any, error) {
+		return map[string]any{"node": m.Node, "capacity": p.Capacity(), "running": p.InHand(), "paused": p.Paused()}, nil
+	}
+	return []stdio.Tool{
+		verb("current", "The Go asks this node runs now, its build capacity, and whether it is paused.", current),
+		verb("pause", "Take no new Go ask on this node until resumed; an ask running here ends in its place. Kept across a restart.",
+			func() (any, error) {
+				if err := p.Pause("the seat's pause verb"); err != nil {
+					return nil, err
+				}
+				return current()
+			}),
+		verb("resume", "Take Go asks again on this node.", func() (any, error) {
+			if err := p.Resume("the seat's resume verb"); err != nil {
+				return nil, err
+			}
+			return current()
+		}),
+	}
+}

modules/mesh-build-go/cmd/mesh-build-go/asks_integration_test.go

@@ -0,0 +1,97 @@
+//go:build integration
+
+package main
+
+import (
+	"context"
+	"os"
+	"sync"
+	"testing"
+	"time"
+
+	"github.com/nats-io/nats.go"
+	"github.com/nats-io/nats.go/jetstream"
+
+	"git.novox.be/novox/mesh-sdk/go/buildstep"
+	"git.novox.be/novox/mesh-sdk/go/delivery"
+)
+
+// 1887/AC-order, 1887/AC-pull against the real bus: a check ask published before a delivery ask runs after it, and at
+// capacity 1 the second waits in the stream until the first ends.
+func TestTheModuleTakesADeliveryAskBeforeAnEarlierCheck(t *testing.T) {
+	url := os.Getenv("MESH_TEST_NATS")
+	if url == "" {
+		t.Skip("MESH_TEST_NATS names no bus: the integration level starts one")
+	}
+	nc, err := nats.Connect(url)
+	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, 30*time.Second); err != nil {
+		t.Fatal(err)
+	}
+	t.Cleanup(func() { _ = js.DeleteStream(context.Background(), "SEAT_NODE_BUILD_GO") })
+	check := deliveryAsk()
+	check.Kind = delivery.AskCheck
+	first := deliveryAsk()
+	for _, a := range []delivery.StepAsk{check, first} {
+		raw, _ := delivery.EncodeStepAsk(a)
+		subject, _ := delivery.AskSubject(a.Language, a.Kind)
+		if _, err := js.Publish(ctx, subject, raw); err != nil {
+			t.Fatal(err)
+		}
+	}
+	states := newStates()
+	m := moduleWith(states, "go1.26.0")
+	m.Capacity = func() int { return 1 }
+	var mu sync.Mutex
+	var order []string
+	steps := fakeSteps(new([]string))
+	for i := range steps {
+		if steps[i].Name == "build_go_fetch" {
+			run := steps[i].Run
+			steps[i].Run = func(ctx context.Context, args map[string]any) (any, error) {
+				mu.Lock()
+				order = append(order, args["commit"].(string))
+				mu.Unlock()
+				return run(ctx, args)
+			}
+		}
+	}
+	p := m.Puller(steps, &buildstep.JetStreamAsks{JS: js, Language: delivery.AskGo})
+	p.PullWait = 200 * time.Millisecond
+	go p.Pull(ctx)
+	for _, a := range []delivery.StepAsk{first, check} {
+		key, _ := delivery.AskStepKey("ace", a.ID)
+		for deadline := time.Now().Add(10 * time.Second); ; time.Sleep(10 * time.Millisecond) {
+			states.mu.Lock()
+			v := states.steps[key]
+			states.mu.Unlock()
+			if len(v) > 0 && v[len(v)-1].State != delivery.StepRunning {
+				break
+			}
+			if time.Now().After(deadline) {
+				t.Fatalf("the ask %s (%s) never ended", a.ID, a.Kind)
+			}
+		}
+		states.mu.Lock()
+		checkKey, _ := delivery.AskStepKey("ace", check.ID)
+		if a.ID == first.ID && len(states.steps[checkKey]) > 0 {
+			states.mu.Unlock()
+			t.Fatal("the check ask started before the delivery ask ended at capacity 1")
+		}
+		states.mu.Unlock()
+	}
+	mu.Lock()
+	defer mu.Unlock()
+	if len(order) != 2 {
+		t.Fatalf("fetched %v", order)
+	}
+}

modules/mesh-build-go/cmd/mesh-build-go/asks_test.go

@@ -0,0 +1,150 @@
+package main
+
+import (
+	"context"
+	"encoding/json"
+	"os"
+	"slices"
+	"strings"
+	"testing"
+	"time"
+
+	"git.novox.be/novox/mesh-sdk/go/buildstep"
+	"git.novox.be/novox/mesh-sdk/go/delivery"
+)
+
+const askCommit = "0123456789abcdef0123456789abcdef01234567"
+
+// fakeSteps are step tools that answer at once: a fetch with a workspace, a build with a digest and a reference.
+func fakeSteps(calls *[]string) []StepTool {
+	answer := func(name string, out any) StepTool {
+		return StepTool{Name: name, Run: func(ctx context.Context, args map[string]any) (any, error) {
+			*calls = append(*calls, name)
+			return out, nil
+		}}
+	}
+	return []StepTool{
+		answer("build_go_fetch", map[string]any{"workspace": "ws1"}),
+		answer("build_go_build", map[string]any{"digest": "sha256:aa", "reference": "artifact-store://mesh-build-go/tools@sha256:aa"}),
+		answer("build_go_digest", map[string]any{"level": "unit", "packages": map[string]string{}}),
+		answer("build_go_test", map[string]any{"passed": true}),
+	}
+}
+
+func deliveryAsk() delivery.StepAsk {
+	return delivery.StepAsk{ID: buildstep.NewID(), Kind: delivery.AskDelivery, Language: delivery.AskGo, Asker: "mesh-delivery-builder",
+		At: time.Now(), Build: &delivery.BuildJob{Module: "mesh-delivery-view", Path: "modules/mesh-delivery-view", Publish: true,
+			Fetch:     map[string]any{"repository": "novox/mesh-delivery", "commit": askCommit},
+			Artifacts: []delivery.AskArtifact{{Artifact: "program"}}}}
+}
+
+// 1887/A3go: an ask pulled runs build_go_build after build_go_fetch, and the step key <node>.<ask id> names the
+// reference.
+func TestAPulledAskRunsTheBuildAndItsKeyNamesTheReference(t *testing.T) {
+	states := newStates()
+	m := moduleWith(states, "go1.26.0")
+	m.Capacity = func() int { return 1 }
+	var calls []string
+	src := buildstep.NewFakeAsks()
+	a := deliveryAsk()
+	if err := src.PublishAsk(a); err != nil {
+		t.Fatal(err)
+	}
+	p := m.Puller(fakeSteps(&calls), src)
+	p.PullWait = 10 * time.Millisecond
+	ctx, stop := context.WithCancel(context.Background())
+	defer stop()
+	go p.Pull(ctx)
+	key, _ := delivery.AskStepKey("ace", a.ID)
+	var step delivery.Step
+	for deadline := time.Now().Add(3 * time.Second); ; time.Sleep(5 * time.Millisecond) {
+		states.mu.Lock()
+		v := states.steps[key]
+		states.mu.Unlock()
+		if len(v) > 0 && v[len(v)-1].State != delivery.StepRunning {
+			step = v[len(v)-1]
+			break
+		}
+		if time.Now().After(deadline) {
+			t.Fatalf("the ask's step %s never ended: %v", key, v)
+		}
+	}
+	if step.State != delivery.StepPassed || !strings.Contains(string(step.Result), "artifact-store://mesh-build-go/tools@sha256:aa") {
+		t.Fatalf("the step = %+v", step)
+	}
+	if strings.Join(calls, ",") != "build_go_fetch,build_go_build" {
+		t.Errorf("calls = %v", calls)
+	}
+	if acked, _, _ := src.Fates(); len(acked) != 1 || acked[0] != a.ID {
+		t.Errorf("acked %v", acked)
+	}
+}
+
+// 1887/AC-zero: at capacity 0 the node's module pulls nothing, and its tools still answer a direct call.
+func TestAtCapacityZeroNothingIsPulledAndTheToolsAnswer(t *testing.T) {
+	states := newStates()
+	m := moduleWith(states, "go1.26.0")
+	m.Capacity = func() int { return 0 }
+	var calls []string
+	steps := fakeSteps(&calls)
+	src := buildstep.NewFakeAsks()
+	_ = src.PublishAsk(deliveryAsk())
+	p := m.Puller(steps, src)
+	p.PullWait = 10 * time.Millisecond
+	ctx, stop := context.WithCancel(context.Background())
+	defer stop()
+	go p.Pull(ctx)
+	time.Sleep(50 * time.Millisecond)
+	if src.Waiting(delivery.AskDelivery) != 1 || len(calls) != 0 {
+		t.Fatalf("a node at capacity 0 pulled: waiting %d, calls %v", src.Waiting(delivery.AskDelivery), calls)
+	}
+	for _, tool := range m.Tools(steps) {
+		if tool.Name == "build_go_fetch" {
+			out, err := tool.Run(map[string]any{"repository": "novox/x", "commit": askCommit})
+			if err != nil || !strings.Contains(string(mustJSON(out)), `"workspace":"ws1"`) || len(calls) != 1 {
+				t.Fatalf("the direct call answered %v, %v", out, err)
+			}
+			return
+		}
+	}
+	t.Fatal("no fetch tool")
+}
+
+func mustJSON(v any) []byte { b, _ := json.Marshal(v); return b }
+
+// 1887/A3go, A7: the manifest claims node-build-go serving its three verbs, reads the checker's kept verdicts and takes
+// the node's build capacity as a setting, with the builder's default; the seat's verbs are served as its tools.
+func TestManifestClaimsTheGoBuildSeatAndReadsItsCapacity(t *testing.T) {
+	m := readManifest(t)
+	if len(m.Claims) != 1 || m.Claims[0].Name != Seat || !slices.Equal(m.Claims[0].Serves, []string{"current", "pause", "resume"}) {
+		t.Fatalf("claims = %+v", m.Claims)
+	}
+	if !slices.Contains(m.Reads, KeptState) {
+		t.Errorf("reads = %v lack %s", m.Reads, KeptState)
+	}
+	raw, _ := os.ReadFile(moduleDir + "/module.json")
+	var doc struct {
+		Config map[string]struct {
+			Default string `json:"default"`
+		} `json:"config"`
+	}
+	_ = json.Unmarshal(raw, &doc)
+	if doc.Config[buildstep.CapacitySetting].Default != buildstep.DefaultCapacities {
+		t.Errorf("the capacity's default = %q, want %q", doc.Config[buildstep.CapacitySetting].Default, buildstep.DefaultCapacities)
+	}
+	settings := resource(t, m, "settings")
+	if settings["path"] != "${dir:state}/settings.json" {
+		t.Errorf("the settings file is at %v", settings["path"])
+	}
+	served := map[string]bool{}
+	for _, tool := range m0().SeatTools(m0().Puller(nil, buildstep.NewFakeAsks())) {
+		served[tool.Name] = true
+	}
+	for _, verb := range m.Claims[0].Serves {
+		if !served[Seat+"."+verb] {
+			t.Errorf("%s.%s is not served", Seat, verb)
+		}
+	}
+}
+
+func m0() *Module { return &Module{Node: "ace", Capacity: func() int { return 0 }} }

modules/mesh-build-go/cmd/mesh-build-go/main.go

@@ -52,6 +52,8 @@ func main() {
 	}
 	m.Run = StepRunner(m.Home)
 	m.RegistryFile = m.Home + "/registry"
+	m.Settings = m.Home + "/settings.json"
+	m.ReadKept = readOf(stdio.State(KeptState)).Get
 	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.
@@ -64,7 +66,13 @@ func main() {
 		logf("announcing the states it keeps: %v", err)
 		os.Exit(1)
 	}
-	if err := stdio.Serve("", m.Tools(m.StepTools())); err != nil {
+	steps := m.StepTools()
+	puller := m.Puller(steps, nil)
+	if m.Node != "" {
+		// The seat's asks, pulled once Serve has the runtime's channel; a pull before it is said and tried again.
+		go func() { _ = puller.Pull(ctx) }()
+	}
+	if err := stdio.Serve("", append(m.Tools(steps), m.SeatTools(puller)...)); err != nil {
 		logf("%v", err)
 		os.Exit(1)
 	}

modules/mesh-build-go/cmd/mesh-build-go/manifest_test.go

@@ -84,9 +84,10 @@ func TestWorkAndCacheDirectoriesLieOutsideTheHome(t *testing.T) {
 	}
 }
 
-// The states are the three the module keeps, no seat is claimed (ADR 0469 §6), and the health is the module's own
-// tool, which the manifest also lists among its tools.
-func TestManifestKeepsItsStatesClaimsNoSeatAndDeclaresItsHealthTool(t *testing.T) {
+// The states are the three the module keeps, the one seat it claims is its language's build seat (ADR 0525 §2, which
+// changes ADR 0469 §6's "no seat"), and the health is the module's own tool, which the manifest also lists among its
+// tools.
+func TestManifestKeepsItsStatesClaimsItsBuildSeatAndDeclaresItsHealthTool(t *testing.T) {
 	m := readManifest(t)
 	states := []string{}
 	for _, s := range m.State {
@@ -96,8 +97,8 @@ func TestManifestKeepsItsStatesClaimsNoSeatAndDeclaresItsHealthTool(t *testing.T
 	if want := []string{"keys", "steps", "toolchains"}; !slices.Equal(states, want) {
 		t.Errorf("state = %v, want %v", states, want)
 	}
-	if len(m.Claims) != 0 {
-		t.Errorf("claims = %v, want no seat", m.Claims)
+	if len(m.Claims) != 1 || m.Claims[0].Name != Seat {
+		t.Errorf("claims = %v, want the seat %s alone", m.Claims, Seat)
 	}
 	if !slices.Contains(m.Tools, HealthTool) {
 		t.Errorf("tools = %v lack %s", m.Tools, HealthTool)

modules/mesh-build-go/cmd/mesh-build-go/module.go

@@ -2,6 +2,7 @@ package main
 
 import (
 	"context"
+	"encoding/json"
 	"fmt"
 	"net/http"
 	"sync"
@@ -76,6 +77,12 @@ type Module struct {
 	// floorOf(node), and a zero Floor checks nothing.
 	Floor buildstep.Floor
 
+	// Settings is the module's settings file, which holds the node's build capacity (buildstep.CapacitySetting); Capacity,
+	// when set, answers it instead (a test). ReadKept reads a key of the checker's kept state (KeptState).
+	Settings string
+	Capacity func() int
+	ReadKept func(key string) (json.RawMessage, error)
+
 	spacesOnce sync.Once
 	workspaces *buildstep.Workspaces
 }

modules/mesh-build-go/files/settings.json

@@ -0,0 +1,3 @@
+{
+  "capacity": "shanks=3,ace=2,g14=0,novox=0"
+}

modules/mesh-build-go/module.json

@@ -12,6 +12,24 @@
     "keys",
     "steps"
   ],
+  "claims": [
+    {
+      "name": "node-build-go",
+      "scope": "node",
+      "serves": ["current", "pause", "resume"]
+    }
+  ],
+  "reads": [
+    "mesh-delivery-checker.kept"
+  ],
+  "config": {
+    "capacity": {
+      "type": "string",
+      "description": "How many asks of the Go build seat this module runs at once on each node, as node=asks pairs separated by commas, such as shanks=3,ace=2 (novox/hq ADR 0525 §5); a node set to 0 pulls nothing, a node not named runs one at a time. The control-node and the laptop build nothing.",
+      "pattern": "|[a-z0-9_-]+=[0-9]+(,[a-z0-9_-]+=[0-9]+)*",
+      "default": "shanks=3,ace=2,g14=0,novox=0"
+    }
+  },
   "tools": [
     "build_go_fetch",
     "build_go_test",
@@ -79,6 +97,17 @@
       "owner": "mesh-build-go",
       "unknown": "program"
     },
+    {
+      "id": "settings",
+      "trusted": false,
+      "type": "file",
+      "path": "${dir:state}/settings.json",
+      "mode": "0600",
+      "owner": "mesh-build-go",
+      "takes-config": ["capacity"],
+      "merge": "json",
+      "template": "files/settings.json"
+    },
     {
       "id": "ssh-dir",
       "type": "directory",