Pull a build seat's asks delivery first, at most the node's capacity, each ended on its own step key, so no part chooses a node or holds a check (issue 1887 A2a, ADR 0525 §4-§6)

2026-10-23 15:15:57 +02:00
parent b9b9c5120c
commit a1d73bdcf0
6 changed files with 767 additions and 31 deletions
@@ -0,0 +1,257 @@
+package buildstep
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"sync"
+	"time"
+
+	"git.novox.be/novox/mesh-sdk/go/delivery"
+)
+
+// **The pull loop the build modules share** (novox/hq ADR 0525 §4-§6, ADR 0536 §4; issue 1887). A build module on a node
+// whose build capacity is above zero pulls the asks of its language's seat: from the delivery worker first, then check,
+// then recheck, holding at most its capacity at once. Each ask runs as a step under the ask's own id, so its end is the
+// key <node>.<ask id> in the module's steps state, which the asker watches; the ask is acknowledged when that end is
+// written, kept alive while it runs by an in-progress signal, and redelivered by the stream to another node if this one
+// stops. No part keeps a timer for an ask: the pull's own wait and the stream's acknowledgement wait are the bus's.
+
+// Pulled is one ask a source handed over, with what decides its fate on the bus.
+type Pulled struct {
+	// Kind is the worker it came from, filled by the puller.
+	Kind string
+	Data []byte
+	// Delivered is how many times the bus has handed it over, this time included.
+	Delivered uint64
+	// Ack ends it: it is never delivered again. InProgress tells the bus it is still being worked, restarting its
+	// acknowledgement wait. Nak hands it back to be delivered again at once (to this node or another). Term drops it.
+	Ack, InProgress, Nak, Term func() error
+}
+
+// AskSource is where a puller takes asks from: the seat's workers on the bus (JetStreamAsks) or a test double
+// (FakeAsks).
+type AskSource interface {
+	// Next takes one ask of kind: with wait 0 only one already waiting, else waiting for one at most wait. It answers
+	// nil and no error when none came.
+	Next(ctx context.Context, kind string, wait time.Duration) (*Pulled, error)
+}
+
+// Defaults of a Puller.
+var (
+	// DefaultPullWait is how long one pull of the delivery worker waits for an ask while every worker is empty: the
+	// pull's own wait (as process.Consume's), after which the check and recheck workers are looked at again.
+	DefaultPullWait = 5 * time.Second
+	// DefaultHeartbeat is how often a running ask tells the bus it is still being worked; under the seat worker's
+	// acknowledgement wait of 60 s (mesh-controller internal/broker HolderConsumerFor).
+	DefaultHeartbeat = 20 * time.Second
+)
+
+// Puller pulls a build seat's asks and runs them.
+type Puller struct {
+	Source   AskSource
+	Language string // the seat's language: delivery.AskGo, AskNode or AskImage
+	Node     string // the node this holder runs on, the first part of its step keys
+	Put      Put    // writes the module's steps state
+	// Capacity is how many asks the node runs at once; read before every pull, so a change takes effect at the next
+	// one. 0 pulls nothing.
+	Capacity func() int
+	// Run runs one ask's job; a *Refusal ends its step refused. Its result is the step's result.
+	Run func(ctx context.Context, ask delivery.StepAsk) (any, error)
+	// PullWait and Heartbeat default to DefaultPullWait and DefaultHeartbeat.
+	PullWait  time.Duration
+	Heartbeat time.Duration
+	Logf      func(format string, args ...any)
+
+	mu     sync.Mutex
+	inHand int
+	paused bool
+	wake   chan struct{}
+	once   sync.Once
+}
+
+func (p *Puller) init() {
+	p.once.Do(func() { p.wake = make(chan struct{}, 1) })
+}
+
+func (p *Puller) logf(format string, args ...any) {
+	if p.Logf != nil {
+		p.Logf(format, args...)
+	}
+}
+
+// Wake makes a waiting puller look again: after a capacity change, a resume, or a step's end.
+func (p *Puller) Wake() {
+	p.init()
+	select {
+	case p.wake <- struct{}{}:
+	default:
+	}
+}
+
+// Pause stops the puller taking asks; the asks it holds run to their end.
+func (p *Puller) Pause() {
+	p.mu.Lock()
+	p.paused = true
+	p.mu.Unlock()
+}
+
+// Resume lets the puller take asks again.
+func (p *Puller) Resume() {
+	p.mu.Lock()
+	p.paused = false
+	p.mu.Unlock()
+	p.Wake()
+}
+
+// Paused says whether the puller takes no asks.
+func (p *Puller) Paused() bool {
+	p.mu.Lock()
+	defer p.mu.Unlock()
+	return p.paused
+}
+
+// InHand is how many asks run now.
+func (p *Puller) InHand() int {
+	p.mu.Lock()
+	defer p.mu.Unlock()
+	return p.inHand
+}
+
+// room says whether the puller may take one more ask now.
+func (p *Puller) room() bool {
+	capacity := 0
+	if p.Capacity != nil {
+		capacity = p.Capacity()
+	}
+	p.mu.Lock()
+	defer p.mu.Unlock()
+	return !p.paused && p.inHand < capacity
+}
+
+// Pull takes and runs asks until ctx ends. An ask still running then is neither ended nor acknowledged: the stream
+// delivers it again, to another node or to this one restarted.
+func (p *Puller) Pull(ctx context.Context) error {
+	p.init()
+	if p.Source == nil || p.Put == nil || p.Run == nil {
+		return errors.New("buildstep: a puller needs its source, its steps' put and its run")
+	}
+	wait := p.PullWait
+	if wait <= 0 {
+		wait = DefaultPullWait
+	}
+	for {
+		if err := ctx.Err(); err != nil {
+			return err
+		}
+		if !p.room() {
+			select {
+			case <-ctx.Done():
+				return ctx.Err()
+			case <-p.wake:
+			}
+			continue
+		}
+		msg, err := p.next(ctx, wait)
+		if err != nil {
+			if ctx.Err() != nil {
+				return ctx.Err()
+			}
+			p.logf("the %s asks cannot be pulled: %v", p.Language, err)
+			select { // the pull's own wait before the next try, as an empty pull's
+			case <-ctx.Done():
+				return ctx.Err()
+			case <-p.wake:
+			case <-time.After(wait):
+			}
+			continue
+		}
+		if msg == nil {
+			continue
+		}
+		p.take(ctx, msg)
+	}
+}
+
+// next is one ask from the workers in their order: one already waiting on delivery, check or recheck, and when none
+// waits, a pull of the delivery worker that waits for one.
+func (p *Puller) next(ctx context.Context, wait time.Duration) (*Pulled, error) {
+	for _, kind := range delivery.AskKinds {
+		msg, err := p.Source.Next(ctx, kind, 0)
+		if err != nil || msg != nil {
+			return withKind(msg, kind), err
+		}
+	}
+	msg, err := p.Source.Next(ctx, delivery.AskDelivery, wait)
+	return withKind(msg, delivery.AskDelivery), err
+}
+
+func withKind(msg *Pulled, kind string) *Pulled {
+	if msg != nil {
+		msg.Kind = kind
+	}
+	return msg
+}
+
+// take starts one ask: it decodes it, refuses one that is not of this seat, and runs it as a step under its own id.
+func (p *Puller) take(ctx context.Context, msg *Pulled) {
+	kind := msg.Kind
+	ask, err := delivery.DecodeStepAsk(msg.Data)
+	if err == nil && (ask.Language != p.Language || (kind != "" && ask.Kind != kind)) {
+		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.
+		p.logf("an ask dropped: %v", err)
+		_ = msg.Term()
+		return
+	}
+	step, key, err := begin(p.Put, p.Node, ask.Kind+" ask", ask.ID)
+	if err != nil {
+		p.logf("the ask %s cannot start on %s: %v", ask.ID, p.Node, err)
+		_ = msg.Nak()
+		return
+	}
+	p.mu.Lock()
+	p.inHand++
+	p.mu.Unlock()
+	go p.run(ctx, msg, ask, key, step)
+}
+
+func (p *Puller) run(ctx context.Context, msg *Pulled, ask delivery.StepAsk, key string, step delivery.Step) {
+	defer func() {
+		p.mu.Lock()
+		p.inHand--
+		p.mu.Unlock()
+		p.Wake()
+	}()
+	beat := p.Heartbeat
+	if beat <= 0 {
+		beat = DefaultHeartbeat
+	}
+	done := make(chan struct{})
+	go func() {
+		t := time.NewTicker(beat)
+		defer t.Stop()
+		for {
+			select {
+			case <-done:
+				return
+			case <-t.C:
+				if msg.InProgress != nil {
+					_ = msg.InProgress()
+				}
+			}
+		}
+	}()
+	result, err := protect(ctx, func(ctx context.Context) (any, error) { return p.Run(ctx, ask) })
+	close(done)
+	if ctx.Err() != nil {
+		// The process stops: the ask is neither ended nor acknowledged here, so the stream gives it to a node that runs.
+		return
+	}
+	finish(p.Put, key, step, result, err)
+	if aerr := msg.Ack(); aerr != nil {
+		p.logf("the ask %s ended on %s and its acknowledgement failed: %v", ask.ID, p.Node, aerr)
+	}
+}
@@ -0,0 +1,102 @@
+package buildstep
+
+import (
+	"context"
+	"fmt"
+	"sync"
+	"time"
+
+	"github.com/nats-io/nats.go/jetstream"
+
+	"git.novox.be/novox/mesh-sdk/go/delivery"
+)
+
+// JetStreamAsks is a build seat's workers on the bus: the shared pull consumer of each kind on the seat's stream
+// (delivery.AskStream, AskWorker), which the broker makes when the seat is declared; a holder only binds them.
+type JetStreamAsks struct {
+	JS       jetstream.JetStream
+	Language string
+
+	mu   sync.Mutex
+	cons map[string]jetstream.Consumer
+}
+
+func (s *JetStreamAsks) consumer(ctx context.Context, kind string) (jetstream.Consumer, error) {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	if c, ok := s.cons[kind]; ok {
+		return c, nil
+	}
+	stream, err := delivery.AskStream(s.Language)
+	if err != nil {
+		return nil, err
+	}
+	worker, err := delivery.AskWorker(s.Language, kind)
+	if err != nil {
+		return nil, err
+	}
+	c, err := s.JS.Consumer(ctx, stream, worker)
+	if err != nil {
+		return nil, fmt.Errorf("binding the %s worker %s on %s: %w", kind, worker, stream, err)
+	}
+	if s.cons == nil {
+		s.cons = map[string]jetstream.Consumer{}
+	}
+	s.cons[kind] = c
+	return c, nil
+}
+
+// Next is AskSource's: a fetch of one with no wait when wait is 0, else one that waits at most wait.
+func (s *JetStreamAsks) Next(ctx context.Context, kind string, wait time.Duration) (*Pulled, error) {
+	c, err := s.consumer(ctx, kind)
+	if err != nil {
+		return nil, err
+	}
+	var batch jetstream.MessageBatch
+	if wait <= 0 {
+		batch, err = c.FetchNoWait(1)
+	} else {
+		batch, err = c.Fetch(1, jetstream.FetchMaxWait(wait))
+	}
+	if err != nil {
+		return nil, err
+	}
+	select {
+	case msg, ok := <-batch.Messages():
+		if !ok {
+			return nil, batch.Error()
+		}
+		var delivered uint64 = 1
+		if meta, err := msg.Metadata(); err == nil && meta != nil {
+			delivered = meta.NumDelivered
+		}
+		return &Pulled{Data: append([]byte(nil), msg.Data()...), Delivered: delivered,
+			Ack: msg.Ack, InProgress: msg.InProgress, Nak: msg.Nak, Term: msg.Term}, nil
+	case <-ctx.Done():
+		return nil, ctx.Err()
+	}
+}
+
+// DeclareAskWorkers makes language's seat stream and its three workers on a bus with no broker: a test's (the build
+// modules' integration tests start one). On the mesh the broker declares them from the seat, as it declares every
+// seat's stream (mesh-controller internal/broker SeatStreams); a holder never does.
+func DeclareAskWorkers(ctx context.Context, js jetstream.JetStream, language string, ackWait time.Duration) error {
+	stream, err := delivery.AskStream(language)
+	if err != nil {
+		return err
+	}
+	seat, _ := delivery.BuildSeat(language)
+	if _, err := js.CreateOrUpdateStream(ctx, jetstream.StreamConfig{Name: stream, Subjects: []string{"mesh.seat." + seat + ".accept.>"},
+		Retention: jetstream.WorkQueuePolicy}); err != nil {
+		return fmt.Errorf("the stream %s: %w", stream, err)
+	}
+	for _, kind := range delivery.AskKinds {
+		worker, _ := delivery.AskWorker(language, kind)
+		subject, _ := delivery.AskSubject(language, kind)
+		if _, err := js.CreateOrUpdateConsumer(ctx, stream, jetstream.ConsumerConfig{Durable: worker, FilterSubject: subject,
+			AckPolicy: jetstream.AckExplicitPolicy, AckWait: ackWait, MaxDeliver: 5}); err != nil {
+			return fmt.Errorf("the worker %s: %w", worker, err)
+		}
+	}
+	return nil
+}
@@ -0,0 +1,107 @@
+package buildstep
+
+import (
+	"context"
+	"sync"
+	"time"
+
+	"git.novox.be/novox/mesh-sdk/go/delivery"
+)
+
+// FakeAsks is a test double of a seat's workers for the build modules' unit tests: Publish puts an ask on a kind's
+// worker, Next hands them out in their order, and the fates the puller gives them are counted. Safe for concurrent use.
+type FakeAsks struct {
+	mu      sync.Mutex
+	waiting map[string][][]byte
+	added   chan struct{}
+	// Acked, Termed and Naked are the asks' ids or data by fate, InProgress how many heartbeats were given.
+	Acked, Termed, Naked []string
+	InProgress           int
+}
+
+// NewFakeAsks is an empty double.
+func NewFakeAsks() *FakeAsks {
+	return &FakeAsks{waiting: map[string][][]byte{}, added: make(chan struct{}, 1)}
+}
+
+// Publish puts an encoded ask on kind's worker.
+func (f *FakeAsks) Publish(kind string, data []byte) {
+	f.mu.Lock()
+	f.waiting[kind] = append(f.waiting[kind], data)
+	f.mu.Unlock()
+	select {
+	case f.added <- struct{}{}:
+	default:
+	}
+}
+
+// PublishAsk encodes a and puts it on its kind's worker.
+func (f *FakeAsks) PublishAsk(a delivery.StepAsk) error {
+	raw, err := delivery.EncodeStepAsk(a)
+	if err != nil {
+		return err
+	}
+	f.Publish(a.Kind, raw)
+	return nil
+}
+
+// Waiting is how many asks wait on kind's worker.
+func (f *FakeAsks) Waiting(kind string) int {
+	f.mu.Lock()
+	defer f.mu.Unlock()
+	return len(f.waiting[kind])
+}
+
+// Fates is a copy of the acked, termed and naked lists.
+func (f *FakeAsks) Fates() (acked, termed, naked []string) {
+	f.mu.Lock()
+	defer f.mu.Unlock()
+	return append([]string(nil), f.Acked...), append([]string(nil), f.Termed...), append([]string(nil), f.Naked...)
+}
+
+// Next is AskSource's.
+func (f *FakeAsks) Next(ctx context.Context, kind string, wait time.Duration) (*Pulled, error) {
+	var deadline <-chan time.Time
+	if wait > 0 {
+		t := time.NewTimer(wait)
+		defer t.Stop()
+		deadline = t.C
+	}
+	for {
+		f.mu.Lock()
+		if q := f.waiting[kind]; len(q) > 0 {
+			data := q[0]
+			f.waiting[kind] = q[1:]
+			f.mu.Unlock()
+			return f.pulled(data), nil
+		}
+		f.mu.Unlock()
+		if wait <= 0 {
+			return nil, nil
+		}
+		select {
+		case <-ctx.Done():
+			return nil, ctx.Err()
+		case <-deadline:
+			return nil, nil
+		case <-f.added:
+		}
+	}
+}
+
+func (f *FakeAsks) pulled(data []byte) *Pulled {
+	id := string(data)
+	if a, err := delivery.DecodeStepAsk(data); err == nil {
+		id = a.ID
+	}
+	fate := func(list *[]string) func() error {
+		return func() error {
+			f.mu.Lock()
+			*list = append(*list, id)
+			f.mu.Unlock()
+			return nil
+		}
+	}
+	return &Pulled{Data: data, Delivered: 1, Ack: fate(&f.Acked), Nak: fate(&f.Naked), Term: fate(&f.Termed),
+		InProgress: func() error { f.mu.Lock(); f.InProgress++; f.mu.Unlock(); return nil }}
+}
@@ -0,0 +1,98 @@
+//go:build integration
+
+package buildstep
+
+import (
+	"context"
+	"sync"
+	"testing"
+	"time"
+
+	"github.com/nats-io/nats.go"
+	"github.com/nats-io/nats.go/jetstream"
+
+	"git.novox.be/novox/mesh-sdk/go/delivery"
+	"git.novox.be/novox/mesh-sdk/go/testbus"
+)
+
+func askBus(t *testing.T, ackWait time.Duration) jetstream.JetStream {
+	t.Helper()
+	nc, err := nats.Connect(testbus.URL(t))
+	if err != nil {
+		t.Fatal(err)
+	}
+	t.Cleanup(nc.Close)
+	js, err := jetstream.New(nc)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if err := DeclareAskWorkers(context.Background(), js, delivery.AskGo, ackWait); err != nil {
+		t.Fatal(err)
+	}
+	return js
+}
+
+func publishAsk(t *testing.T, js jetstream.JetStream, a delivery.StepAsk) {
+	t.Helper()
+	raw, err := delivery.EncodeStepAsk(a)
+	if err != nil {
+		t.Fatal(err)
+	}
+	subject, _ := delivery.AskSubject(a.Language, a.Kind)
+	if _, err := js.Publish(context.Background(), subject, raw); err != nil {
+		t.Fatal(err)
+	}
+}
+
+// 1887/A2a, 1887/AC-order, 1887/AC-pull on the real server: a check ask published before a delivery ask runs after it;
+// with capacity 2 and three asks two run, the third waits in the stream and starts when one ends.
+func TestPullOrderAndCapacityOnTheBus(t *testing.T) {
+	js := askBus(t, 30*time.Second)
+	check, first, second := testAsk(delivery.AskCheck), testAsk(delivery.AskDelivery), testAsk(delivery.AskDelivery)
+	publishAsk(t, js, check)
+	publishAsk(t, js, first)
+	publishAsk(t, js, second)
+	release := make(chan struct{})
+	var mu sync.Mutex
+	var order []string
+	st := newSteps()
+	p := &Puller{Source: &JetStreamAsks{JS: js, Language: delivery.AskGo}, Language: delivery.AskGo, Node: "shanks", Put: st.put,
+		Capacity: func() int { return 2 }, PullWait: 200 * time.Millisecond,
+		Run: func(ctx context.Context, a delivery.StepAsk) (any, error) {
+			mu.Lock()
+			order = append(order, a.ID)
+			mu.Unlock()
+			<-release
+			return nil, nil
+		}}
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+	go p.Pull(ctx)
+	waitFor(t, "two in hand", func() bool { return p.InHand() == 2 })
+	time.Sleep(500 * time.Millisecond)
+	mu.Lock()
+	if len(order) != 2 || order[0] != first.ID || order[1] != second.ID {
+		t.Fatalf("ran %v, want the two delivery asks %s %s first", order, first.ID, second.ID)
+	}
+	mu.Unlock()
+	release <- struct{}{}
+	waitFor(t, "the check taken", func() bool { mu.Lock(); defer mu.Unlock(); return len(order) == 3 })
+	close(release)
+	waitFor(t, "every step ended", func() bool {
+		for _, a := range []delivery.StepAsk{check, first, second} {
+			key, _ := delivery.AskStepKey("shanks", a.ID)
+			if s, ok := st.get(key); !ok || s.State != delivery.StepPassed {
+				return false
+			}
+		}
+		return true
+	})
+	info, err := js.Stream(ctx, "SEAT_NODE_BUILD_GO")
+	if err != nil {
+		t.Fatal(err)
+	}
+	waitFor(t, "the stream empty (every ask acknowledged)", func() bool {
+		si, err := info.Info(ctx)
+		return err == nil && si.State.Msgs == 0
+	})
+}
@@ -0,0 +1,155 @@
+package buildstep
+
+import (
+	"context"
+	"encoding/json"
+	"sync"
+	"testing"
+	"time"
+
+	"git.novox.be/novox/mesh-sdk/go/delivery"
+)
+
+// steps records a steps state's puts: the last value of each key.
+type steps struct {
+	mu   sync.Mutex
+	keys map[string]delivery.Step
+}
+
+func newSteps() *steps { return &steps{keys: map[string]delivery.Step{}} }
+
+func (s *steps) put(key string, value any) (uint64, error) {
+	raw, _ := value.(json.RawMessage)
+	st, err := delivery.DecodeStep(raw)
+	if err != nil {
+		return 0, err
+	}
+	s.mu.Lock()
+	s.keys[key] = st
+	s.mu.Unlock()
+	return 1, nil
+}
+
+func (s *steps) get(key string) (delivery.Step, bool) {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	st, ok := s.keys[key]
+	return st, ok
+}
+
+func testAsk(kind string) delivery.StepAsk {
+	id := NewID()
+	a := delivery.StepAsk{ID: id, Kind: kind, Language: delivery.AskGo, Asker: "mesh-delivery-builder", At: time.Now()}
+	fetch := map[string]any{"repository": "novox/x", "commit": "0123456789abcdef0123456789abcdef01234567"}
+	if kind == delivery.AskDelivery {
+		b := buildJobFixture
+		b.Fetch = fetch
+		a.Build = &b
+	} else {
+		a.Check = &delivery.CheckJob{Fetch: fetch, Level: delivery.LevelUnit}
+	}
+	return a
+}
+
+var buildJobFixture = delivery.BuildJob{Module: "m", Path: ".", Artifacts: []delivery.AskArtifact{{Artifact: "bundle"}}}
+
+func waitFor(t *testing.T, what string, cond func() bool) {
+	t.Helper()
+	deadline := time.Now().Add(2 * time.Second)
+	for !cond() {
+		if time.Now().After(deadline) {
+			t.Fatalf("waited 2 s for %s", what)
+		}
+		time.Sleep(time.Millisecond)
+	}
+}
+
+// 1887/A2a: given a check ask published before a delivery ask and one slot, the delivery ask runs first.
+func TestPullTakesDeliveryBeforeCheck(t *testing.T) {
+	src, st := NewFakeAsks(), newSteps()
+	check, recheck, deliver := testAsk(delivery.AskCheck), testAsk(delivery.AskRecheck), testAsk(delivery.AskDelivery)
+	for _, a := range []delivery.StepAsk{recheck, check, deliver} {
+		if err := src.PublishAsk(a); err != nil {
+			t.Fatal(err)
+		}
+	}
+	var mu sync.Mutex
+	var order []string
+	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) {
+			mu.Lock()
+			order = append(order, a.Kind)
+			mu.Unlock()
+			return map[string]string{"ran": a.ID}, nil
+		}}
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+	go p.Pull(ctx)
+	waitFor(t, "three asks acked", func() bool { a, _, _ := src.Fates(); return len(a) == 3 })
+	mu.Lock()
+	defer mu.Unlock()
+	if want := []string{delivery.AskDelivery, delivery.AskCheck, delivery.AskRecheck}; len(order) != 3 || order[0] != want[0] || order[1] != want[1] || order[2] != want[2] {
+		t.Fatalf("order = %v, want %v", order, want)
+	}
+	key, _ := delivery.AskStepKey("shanks", deliver.ID)
+	s, ok := st.get(key)
+	if !ok || s.State != delivery.StepPassed || string(s.Result) != `{"ran":"`+deliver.ID+`"}` {
+		t.Fatalf("the step %s = %+v (%v)", key, s, ok)
+	}
+}
+
+// 1887/A2a, 1887/AC-pull: given capacity 2 and three asks waiting, two run and the third waits, and it starts when
+// one ends, with no timer.
+func TestPullHoldsAtMostItsCapacity(t *testing.T) {
+	src, st := NewFakeAsks(), newSteps()
+	for i := 0; i < 3; i++ {
+		_ = src.PublishAsk(testAsk(delivery.AskDelivery))
+	}
+	release := make(chan struct{})
+	var mu sync.Mutex
+	started := 0
+	p := &Puller{Source: src, Language: delivery.AskGo, Node: "shanks", Put: st.put, Capacity: func() int { return 2 },
+		PullWait: 10 * time.Millisecond, Run: func(ctx context.Context, a delivery.StepAsk) (any, error) {
+			mu.Lock()
+			started++
+			mu.Unlock()
+			<-release
+			return nil, nil
+		}}
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+	go p.Pull(ctx)
+	waitFor(t, "two in hand", func() bool { return p.InHand() == 2 })
+	time.Sleep(30 * time.Millisecond) // three pull waits: the third must still wait
+	if p.InHand() != 2 || src.Waiting(delivery.AskDelivery) != 1 {
+		t.Fatalf("in hand %d, waiting %d; want 2 and 1", p.InHand(), src.Waiting(delivery.AskDelivery))
+	}
+	release <- struct{}{}
+	waitFor(t, "the third taken", func() bool { return src.Waiting(delivery.AskDelivery) == 0 })
+	mu.Lock()
+	if started != 3 {
+		t.Errorf("started %d, want 3", started)
+	}
+	mu.Unlock()
+	close(release)
+	waitFor(t, "all acked", func() bool { a, _, _ := src.Fates(); return len(a) == 3 })
+}
+
+// An ask that is not one, or not this seat's, is dropped and runs nowhere.
+func TestPullDropsWhatIsNoAsk(t *testing.T) {
+	src, st := NewFakeAsks(), newSteps()
+	src.Publish(delivery.AskDelivery, []byte(`{"v":1}`))
+	other := testAsk(delivery.AskDelivery)
+	other.Language = delivery.AskNode
+	_ = src.PublishAsk(other)
+	ran := false
+	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) { ran = true; return nil, nil }}
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+	go p.Pull(ctx)
+	waitFor(t, "two termed", func() bool { _, tm, _ := src.Fates(); return len(tm) == 2 })
+	if ran {
+		t.Fatal("a wrong ask ran")
+	}
+}
@@ -42,43 +42,60 @@ func NewID() string {
 // the error's text, or refused when the error is a *Refusal; a panic in run ends the step failed, naming it, and
 // the process goes on. If the running write fails, nothing was started and Start answers the error.
 func Start(ctx context.Context, put Put, node, tool string, run func(ctx context.Context) (result any, err error)) (string, error) {
-	step := delivery.Step{ID: NewID(), Tool: tool, Node: node, State: delivery.StepRunning, Started: now().UTC()}
-	key, err := delivery.StepKey(node, step.ID)
+	step, key, err := begin(put, node, tool, NewID())
 	if err != nil {
 		return "", err
 	}
-	if err := write(put, key, step); err != nil {
-		return "", fmt.Errorf("buildstep: writing the running step %s: %w", key, err)
-	}
-	go func() {
-		result, err := protect(ctx, run)
-		step.Ended = now().UTC()
-		switch ref := (*Refusal)(nil); {
-		case err == nil:
-			step.State = delivery.StepPassed
-			if result != nil {
-				raw, merr := json.Marshal(result)
-				if merr != nil {
-					step.State, step.Reason = delivery.StepFailed, "the result is not JSON: "+merr.Error()
-				} else {
-					step.Result = raw
-				}
-			}
-		case errors.As(err, &ref):
-			step.State, step.Reason = delivery.StepRefused, err.Error()
-		default:
-			step.State, step.Reason = delivery.StepFailed, err.Error()
-		}
-		for attempt := 1; ; attempt++ {
-			if werr := write(put, key, step); werr == nil || attempt >= EndAttempts {
-				return
-			}
-			time.Sleep(EndRetry)
-		}
-	}()
+	go end(ctx, put, key, step, run)
 	return step.ID, nil
 }

+// begin writes the step running under id.
+func begin(put Put, node, tool, id string) (delivery.Step, string, error) {
+	step := delivery.Step{ID: id, Tool: tool, Node: node, State: delivery.StepRunning, Started: now().UTC()}
+	key, err := delivery.StepKey(node, step.ID)
+	if err != nil {
+		return delivery.Step{}, "", err
+	}
+	if err := write(put, key, step); err != nil {
+		return delivery.Step{}, "", fmt.Errorf("buildstep: writing the running step %s: %w", key, err)
+	}
+	return step, key, nil
+}
+
+// end runs run and writes the step's end.
+func end(ctx context.Context, put Put, key string, step delivery.Step, run func(ctx context.Context) (any, error)) {
+	result, err := protect(ctx, run)
+	finish(put, key, step, result, err)
+}
+
+// finish writes the step's end from run's result or error: passed, failed or refused (a *Refusal).
+func finish(put Put, key string, step delivery.Step, result any, err error) {
+	step.Ended = now().UTC()
+	switch ref := (*Refusal)(nil); {
+	case err == nil:
+		step.State = delivery.StepPassed
+		if result != nil {
+			raw, merr := json.Marshal(result)
+			if merr != nil {
+				step.State, step.Reason = delivery.StepFailed, "the result is not JSON: "+merr.Error()
+			} else {
+				step.Result = raw
+			}
+		}
+	case errors.As(err, &ref):
+		step.State, step.Reason = delivery.StepRefused, err.Error()
+	default:
+		step.State, step.Reason = delivery.StepFailed, err.Error()
+	}
+	for attempt := 1; ; attempt++ {
+		if werr := write(put, key, step); werr == nil || attempt >= EndAttempts {
+			return
+		}
+		time.Sleep(EndRetry)
+	}
+}
+
 // protect is run with a panic turned into the error that ends the step.
 func protect(ctx context.Context, run func(ctx context.Context) (any, error)) (result any, err error) {
 	defer func() {