Serve a build seat's current, pause and resume from the SDK, naming the asks running, so the three build modules hold their seats one way (issue 1887 A3node, A9a, ADR 0525 §10)

2026-10-23 15:38:03 +02:00
parent e1dca0a39f
commit 1672f40233
3 changed files with 128 additions and 5 deletions
@@ -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()
 	}()
@@ -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): `<seat>.current` (what the node runs, its capacity, whether it is paused), `<seat>.pause` and
+// `<seat>.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()
+		}),
+	}
+}
@@ -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 <seat>.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 })
+}