diff --git a/modules/mesh-build-go/cmd/mesh-build-go/asks.go b/modules/mesh-build-go/cmd/mesh-build-go/asks.go new file mode 100644 index 00000000..3c0ff49a --- /dev/null +++ b/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 . 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() + }), + } +} diff --git a/modules/mesh-build-go/cmd/mesh-build-go/asks_integration_test.go b/modules/mesh-build-go/cmd/mesh-build-go/asks_integration_test.go new file mode 100644 index 00000000..1d185ceb --- /dev/null +++ b/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) + } +} diff --git a/modules/mesh-build-go/cmd/mesh-build-go/asks_test.go b/modules/mesh-build-go/cmd/mesh-build-go/asks_test.go new file mode 100644 index 00000000..c9d34364 --- /dev/null +++ b/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 . 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 }} } diff --git a/modules/mesh-build-go/cmd/mesh-build-go/main.go b/modules/mesh-build-go/cmd/mesh-build-go/main.go index 85296b47..3becd9d0 100644 --- a/modules/mesh-build-go/cmd/mesh-build-go/main.go +++ b/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) } diff --git a/modules/mesh-build-go/cmd/mesh-build-go/manifest_test.go b/modules/mesh-build-go/cmd/mesh-build-go/manifest_test.go index 2791451e..ace29333 100644 --- a/modules/mesh-build-go/cmd/mesh-build-go/manifest_test.go +++ b/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) diff --git a/modules/mesh-build-go/cmd/mesh-build-go/module.go b/modules/mesh-build-go/cmd/mesh-build-go/module.go index 0f306017..3db6e3fb 100644 --- a/modules/mesh-build-go/cmd/mesh-build-go/module.go +++ b/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 } diff --git a/modules/mesh-build-go/files/settings.json b/modules/mesh-build-go/files/settings.json new file mode 100644 index 00000000..3175fe39 --- /dev/null +++ b/modules/mesh-build-go/files/settings.json @@ -0,0 +1,3 @@ +{ + "capacity": "shanks=3,ace=2,g14=0,novox=0" +} diff --git a/modules/mesh-build-go/module.json b/modules/mesh-build-go/module.json index b78bc91a..db0dbe1c 100644 --- a/modules/mesh-build-go/module.json +++ b/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",