From 1672f402338c0218d89b061a5ebb5b9dd4251bd5 Mon Sep 17 00:00:00 2001 From: jochen Date: Fri, 23 Oct 2026 15:38:03 +0200 Subject: [PATCH] =?UTF-8?q?Serve=20a=20build=20seat's=20current,=20pause?= =?UTF-8?q?=20and=20resume=20from=20the=20SDK,=20naming=20the=20asks=20run?= =?UTF-8?q?ning,=20so=20the=20three=20build=20modules=20hold=20their=20sea?= =?UTF-8?q?ts=20one=20way=20(issue=201887=20A3node,=20A9a,=20ADR=200525=20?= =?UTF-8?q?=C2=A710)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- go/buildstep/pull.go | 41 ++++++++++++++++++++++---- go/buildstep/seat_tools.go | 40 +++++++++++++++++++++++++ go/buildstep/seat_tools_test.go | 52 +++++++++++++++++++++++++++++++++ 3 files changed, 128 insertions(+), 5 deletions(-) create mode 100644 go/buildstep/seat_tools.go create mode 100644 go/buildstep/seat_tools_test.go diff --git a/go/buildstep/pull.go b/go/buildstep/pull.go index 5db95d9..32f243e 100644 --- a/go/buildstep/pull.go +++ b/go/buildstep/pull.go @@ -5,6 +5,7 @@ import ( "encoding/json" "errors" "fmt" + "sort" "sync" "time" @@ -68,11 +69,12 @@ type Puller struct { // a restart. Load func(key string) ([]byte, error) - mu sync.Mutex - inHand int - paused bool - wake chan struct{} - once sync.Once + mu sync.Mutex + inHand int + running map[string]RunningAsk + paused bool + wake chan struct{} + once sync.Once } func (p *Puller) init() { @@ -162,6 +164,26 @@ func (p *Puller) InHand() int { return p.inHand } +// RunningAsk is an ask a puller runs now, as its seat's current verb says it. +type RunningAsk struct { + ID string `json:"id"` + Kind string `json:"kind"` + Module string `json:"module,omitempty"` + Started time.Time `json:"started"` +} + +// Running are the asks the puller runs now, oldest first. +func (p *Puller) Running() []RunningAsk { + p.mu.Lock() + defer p.mu.Unlock() + out := make([]RunningAsk, 0, len(p.running)) + for _, r := range p.running { + out = append(out, r) + } + sort.Slice(out, func(i, j int) bool { return out[i].Started.Before(out[j].Started) }) + return out +} + // room says whether the puller may take one more ask now. func (p *Puller) room() bool { capacity := 0 @@ -259,8 +281,16 @@ func (p *Puller) take(ctx context.Context, msg *Pulled) { _ = msg.Nak() return } + module := "" + if ask.Build != nil { + module = ask.Build.Module + } p.mu.Lock() p.inHand++ + if p.running == nil { + p.running = map[string]RunningAsk{} + } + p.running[ask.ID] = RunningAsk{ID: ask.ID, Kind: ask.Kind, Module: module, Started: step.Started} p.mu.Unlock() go p.run(ctx, msg, ask, key, step) } @@ -269,6 +299,7 @@ func (p *Puller) run(ctx context.Context, msg *Pulled, ask delivery.StepAsk, key defer func() { p.mu.Lock() p.inHand-- + delete(p.running, ask.ID) p.mu.Unlock() p.Wake() }() diff --git a/go/buildstep/seat_tools.go b/go/buildstep/seat_tools.go new file mode 100644 index 0000000..b15f1b5 --- /dev/null +++ b/go/buildstep/seat_tools.go @@ -0,0 +1,40 @@ +package buildstep + +import ( + stdio "git.novox.be/novox/mesh-sdk/go" +) + +// SeatTools are a build seat's verbs on one node, as its holder serves them (novox/hq ADR 0525 §10; mesh-sdk catalogue +// buildSeatVerbs): `.current` (what the node runs, its capacity, whether it is paused), `.pause` and +// `.resume`, the pause kept on the node's pause key by the puller. what names the asks in the descriptions ("Go"). +func SeatTools(seat, what string, p *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) { + capacity := 0 + if p.Capacity != nil { + capacity = p.Capacity() + } + return map[string]any{"node": p.Node, "capacity": capacity, "running": p.Running(), "paused": p.Paused()}, nil + } + return []stdio.Tool{ + verb("current", "The "+what+" asks this node runs now (each one's id, kind and when it started), its build capacity, "+ + "and whether it is paused.", current), + verb("pause", "Take no new "+what+" ask on this node until resumed; an ask running here ends in its place. Kept "+ + "across a restart (the node's pause key).", func() (any, error) { + if err := p.Pause("the seat's pause verb"); err != nil { + return nil, err + } + return current() + }), + verb("resume", "Take "+what+" 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/go/buildstep/seat_tools_test.go b/go/buildstep/seat_tools_test.go new file mode 100644 index 0000000..471ec3d --- /dev/null +++ b/go/buildstep/seat_tools_test.go @@ -0,0 +1,52 @@ +package buildstep + +import ( + "context" + "testing" + "time" + + "git.novox.be/novox/mesh-sdk/go/delivery" +) + +// 1887/A9a, ADR 0525 §10: a build seat's verbs are .current, .pause and .resume; current names the ask running, +// pause stops the puller taking asks, resume lets it take them again. +func TestSeatToolsCurrentPauseResume(t *testing.T) { + src, st := NewFakeAsks(), newSteps() + release := make(chan struct{}) + p := &Puller{Source: src, Language: delivery.AskGo, Node: "shanks", Put: st.put, Capacity: func() int { return 1 }, + PullWait: 10 * time.Millisecond, Run: func(ctx context.Context, a delivery.StepAsk) (any, error) { <-release; return nil, nil }} + tools := map[string]func(map[string]any) (any, error){} + for _, tool := range SeatTools("node-build-go", "Go", p) { + tools[tool.Name] = tool.Run + } + for _, name := range []string{"node-build-go.current", "node-build-go.pause", "node-build-go.resume"} { + if tools[name] == nil { + t.Fatalf("no %s", name) + } + } + a := testAsk(delivery.AskDelivery) + _ = src.PublishAsk(a) + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + go p.Pull(ctx) + waitFor(t, "the ask running", func() bool { return p.InHand() == 1 }) + out, _ := tools["node-build-go.current"](nil) + cur := out.(map[string]any) + running := cur["running"].([]RunningAsk) + if len(running) != 1 || running[0].ID != a.ID || running[0].Module != "m" || cur["capacity"] != 1 || cur["paused"] != false { + t.Fatalf("current = %+v", cur) + } + if out, err := tools["node-build-go.pause"](nil); err != nil || out.(map[string]any)["paused"] != true { + t.Fatalf("pause = %v, %v", out, err) + } + close(release) + _ = src.PublishAsk(testAsk(delivery.AskDelivery)) + time.Sleep(50 * time.Millisecond) + if src.Waiting(delivery.AskDelivery) != 1 { + t.Fatal("a paused node took an ask") + } + if _, err := tools["node-build-go.resume"](nil); err != nil { + t.Fatal(err) + } + waitFor(t, "the ask taken after the resume", func() bool { return src.Waiting(delivery.AskDelivery) == 0 }) +}