Put an ask on its build seat's stream and read its step, and name a building key's asks, so the builder and the checker ask no node and a restarted builder never asks twice (issue 1887 A4a, A4b)

go/delivery/builder.go

@@ -184,6 +184,10 @@ type BuilderBuild struct {
 	// does not yet hold: the key is not claimed until that package's trunk key is done and published (novox/hq ADR 0500
 	// clause 6). A reader that does not know the field reads the key as queued.
 	Waits string `json:"waits,omitempty"`
+	// Asks are, on a building key, the ids of the asks the builder put for it (novox/hq ADR 0525 §1; issue 1887 A4b): a
+	// builder that takes the key over at its start waits for those asks' steps and never puts them again. A reader that
+	// does not know the field reads the key as before.
+	Asks []string `json:"asks,omitempty"`
 }
 
 func (b BuilderBuild) check() error {
@@ -216,6 +220,14 @@ func (b BuilderBuild) check() error {
 			return fmt.Errorf("delivery: the reference %q of the %s %q %s", a.Reference, a.Kind, a.Name, p)
 		}
 	}
+	for _, id := range b.Asks {
+		if b.State != BuildBuilding {
+			return fmt.Errorf("delivery: a %s build names the ask %s; only a building key does", b.State, id)
+		}
+		if err := checkAskID(id); err != nil {
+			return err
+		}
+	}
 	if b.Waits != "" {
 		if b.State != BuildQueued {
 			return fmt.Errorf("delivery: a %s build waits for %q; only a queued key waits for a package", b.State, b.Waits)

go/delivery/buildrun/asks.go

@@ -0,0 +1,220 @@
+package buildrun
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"sync"
+	"time"
+
+	"github.com/nats-io/nats.go"
+	"github.com/nats-io/nats.go/jetstream"
+
+	"git.novox.be/novox/mesh-sdk/go/delivery"
+)
+
+// **An asker puts an ask on its build seat's stream and waits for its step** (novox/hq ADR 0525 §1, §6; issue 1887 A4,
+// A5). It chooses no node: whichever build node pulls the ask writes `<node>.<ask id>` in its build module's steps state,
+// and the asker reads the first such key that is not running. The builder and the checker share it.
+
+// AskPublisher puts an encoded ask on subject, de-duplicated by its id.
+type AskPublisher interface {
+	PublishAsk(ctx context.Context, subject, id string, data []byte) error
+}
+
+// AskWaiter waits for an ask's step: the first key `*.<ask id>` of the build module's steps state whose state is not
+// running. A watch, not a poll; it returns when the step ends or ctx does.
+type AskWaiter interface {
+	WaitAsk(ctx context.Context, module, askID string) (delivery.Step, error)
+}
+
+// Asker puts asks and waits for their steps.
+type Asker struct {
+	Publisher AskPublisher
+	Waiter    AskWaiter
+	// Bound is how long the asker waits for an ask's step before it stops waiting; zero waits as long as ctx.
+	Bound time.Duration
+}
+
+// Put puts a on its stream: the seat's accept subject of its language and kind.
+func (k Asker) Put(ctx context.Context, a delivery.StepAsk) error {
+	raw, err := delivery.EncodeStepAsk(a)
+	if err != nil {
+		return err
+	}
+	subject, err := delivery.AskSubject(a.Language, a.Kind)
+	if err != nil {
+		return err
+	}
+	if err := k.Publisher.PublishAsk(ctx, subject, a.ID, raw); err != nil {
+		return fmt.Errorf("putting the %s ask %s: %w", a.Kind, a.ID, err)
+	}
+	return nil
+}
+
+// Wait waits for the step of an ask already put (a builder that started again takes over its own asks with it, never
+// putting them twice) and answers it when it ended, whatever its state.
+func (k Asker) Wait(ctx context.Context, a delivery.StepAsk) (delivery.Step, error) {
+	module, err := delivery.BuildModuleOf(a.Language)
+	if err != nil {
+		return delivery.Step{}, err
+	}
+	wait, cancel := ctx, context.CancelFunc(func() {})
+	if k.Bound > 0 {
+		wait, cancel = context.WithTimeoutCause(ctx, k.Bound, fmt.Errorf("the ask %s's step did not end within %s", a.ID, k.Bound))
+	}
+	defer cancel()
+	step, err := k.Waiter.WaitAsk(wait, module, a.ID)
+	if err != nil {
+		if cause := context.Cause(wait); cause != nil && ctx.Err() == nil {
+			err = cause
+		}
+		return delivery.Step{}, err
+	}
+	return step, nil
+}
+
+// Ask puts a and waits for its step; a step that did not pass is an error naming its node, its state and its reason.
+func (k Asker) Ask(ctx context.Context, a delivery.StepAsk) (delivery.Step, error) {
+	if err := k.Put(ctx, a); err != nil {
+		return delivery.Step{}, err
+	}
+	return k.Passed(k.Wait(ctx, a))
+}
+
+// Passed is step when it passed, else an error naming where it ended and why.
+func (k Asker) Passed(step delivery.Step, err error) (delivery.Step, error) {
+	if err != nil {
+		return step, err
+	}
+	if step.State != delivery.StepPassed {
+		return step, fmt.Errorf("the ask %s on %s %s: %s", step.ID, step.Node, step.State, step.Reason)
+	}
+	return step, nil
+}
+
+// JetStreamAsks is the bus side of an Asker: a JetStream publish de-duplicated by the ask's id (Nats-Msg-Id), and a watch
+// of `*.<ask id>` in the build module's steps bucket.
+type JetStreamAsks struct {
+	JS jetstream.JetStream
+}
+
+// PublishAsk is AskPublisher's.
+func (s JetStreamAsks) PublishAsk(ctx context.Context, subject, id string, data []byte) error {
+	msg := &nats.Msg{Subject: subject, Data: data, Header: nats.Header{}}
+	msg.Header.Set(jetstream.MsgIDHeader, id)
+	_, err := s.JS.PublishMsg(ctx, msg)
+	return err
+}
+
+// WaitAsk is AskWaiter's: the watch hands over what the keys hold now, so a step that ended before it began is read at
+// once, and every change after.
+func (s JetStreamAsks) WaitAsk(ctx context.Context, module, askID string) (delivery.Step, error) {
+	pattern, err := delivery.AskStepWatch(askID)
+	if err != nil {
+		return delivery.Step{}, err
+	}
+	kv, err := s.JS.KeyValue(ctx, delivery.StepsBucket(module))
+	if err != nil {
+		return delivery.Step{}, fmt.Errorf("the bucket %s: %w", delivery.StepsBucket(module), err)
+	}
+	w, err := kv.Watch(ctx, pattern)
+	if err != nil {
+		return delivery.Step{}, err
+	}
+	defer w.Stop()
+	for {
+		select {
+		case <-ctx.Done():
+			return delivery.Step{}, ctx.Err()
+		case e, ok := <-w.Updates():
+			if !ok {
+				return delivery.Step{}, errors.New("the watch of the ask's step ended")
+			}
+			if e == nil || e.Operation() != jetstream.KeyValuePut {
+				continue
+			}
+			step, err := delivery.DecodeStep(e.Value())
+			if err != nil {
+				continue // not a step (a node's pause key never matches an ask id; anything else is not this ask's)
+			}
+			if step.State != delivery.StepRunning {
+				return step, nil
+			}
+		}
+	}
+}
+
+// FakeAsks is a test double of the bus for an asker's tests: each ask put is kept and, when Answer is set, its step is
+// ended at once with what Answer says a build node wrote. Safe for concurrent use.
+type FakeAsks struct {
+	Answer func(a delivery.StepAsk) delivery.Step
+
+	mu    sync.Mutex
+	asks  []delivery.StepAsk
+	steps map[string]delivery.Step
+	waits map[string]chan struct{}
+}
+
+// PublishAsk is AskPublisher's.
+func (f *FakeAsks) PublishAsk(_ context.Context, _, id string, data []byte) error {
+	a, err := delivery.DecodeStepAsk(data)
+	if err != nil {
+		return err
+	}
+	f.mu.Lock()
+	f.asks = append(f.asks, a)
+	f.mu.Unlock()
+	if f.Answer != nil {
+		f.End(f.Answer(a))
+	}
+	return nil
+}
+
+// End writes an ask's step end, as a build node does.
+func (f *FakeAsks) End(step delivery.Step) {
+	f.mu.Lock()
+	if f.steps == nil {
+		f.steps, f.waits = map[string]delivery.Step{}, map[string]chan struct{}{}
+	}
+	f.steps[step.ID] = step
+	ch := f.waits[step.ID]
+	delete(f.waits, step.ID)
+	f.mu.Unlock()
+	if ch != nil {
+		close(ch)
+	}
+}
+
+// WaitAsk is AskWaiter's.
+func (f *FakeAsks) WaitAsk(ctx context.Context, _ string, askID string) (delivery.Step, error) {
+	f.mu.Lock()
+	if f.steps == nil {
+		f.steps, f.waits = map[string]delivery.Step{}, map[string]chan struct{}{}
+	}
+	if s, ok := f.steps[askID]; ok {
+		f.mu.Unlock()
+		return s, nil
+	}
+	ch := f.waits[askID]
+	if ch == nil {
+		ch = make(chan struct{})
+		f.waits[askID] = ch
+	}
+	f.mu.Unlock()
+	select {
+	case <-ctx.Done():
+		return delivery.Step{}, ctx.Err()
+	case <-ch:
+	}
+	f.mu.Lock()
+	defer f.mu.Unlock()
+	return f.steps[askID], nil
+}
+
+// Asks are the asks put so far.
+func (f *FakeAsks) Asks() []delivery.StepAsk {
+	f.mu.Lock()
+	defer f.mu.Unlock()
+	return append([]delivery.StepAsk(nil), f.asks...)
+}

go/delivery/buildrun/asks_integration_test.go

@@ -0,0 +1,61 @@
+//go:build integration
+
+package buildrun
+
+import (
+	"context"
+	"encoding/json"
+	"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"
+)
+
+// 1887/A4a, 1887/AC3 on the real bus: an ask put twice under one id is one message, and the asker reads the step a node
+// wrote under `<node>.<ask id>`.
+func TestAnAskIsPutOnceAndItsStepIsRead(t *testing.T) {
+	nc, err := nats.Connect(testbus.URL(t))
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer nc.Close()
+	js, _ := jetstream.New(nc)
+	ctx := context.Background()
+	if _, err := js.CreateStream(ctx, jetstream.StreamConfig{Name: "SEAT_NODE_BUILD_GO", Subjects: []string{"mesh.seat.node-build-go.accept.>"},
+		Retention: jetstream.WorkQueuePolicy}); err != nil {
+		t.Fatal(err)
+	}
+	kv, err := js.CreateKeyValue(ctx, jetstream.KeyValueConfig{Bucket: delivery.StepsBucket("mesh-build-go")})
+	if err != nil {
+		t.Fatal(err)
+	}
+	bus := JetStreamAsks{JS: js}
+	k := Asker{Publisher: bus, Waiter: bus, Bound: 5 * time.Second}
+	a := goAsk("mvl0frv18d1960")
+	for i := 0; i < 2; i++ {
+		if err := k.Put(ctx, a); err != nil {
+			t.Fatal(err)
+		}
+	}
+	s, _ := js.Stream(ctx, "SEAT_NODE_BUILD_GO")
+	if info, _ := s.Info(ctx); info.State.Msgs != 1 {
+		t.Fatalf("%d messages for one ask put twice", info.State.Msgs)
+	}
+	go func() {
+		time.Sleep(100 * time.Millisecond)
+		key, _ := delivery.AskStepKey("ace", a.ID)
+		running, _ := delivery.EncodeStep(delivery.Step{ID: a.ID, Tool: "delivery ask", Node: "ace", State: delivery.StepRunning, Started: time.Now()})
+		_, _ = kv.Put(ctx, key, running)
+		passed, _ := delivery.EncodeStep(delivery.Step{ID: a.ID, Tool: "delivery ask", Node: "ace", State: delivery.StepPassed, Started: time.Now(),
+			Result: json.RawMessage(`{"artifacts":[]}`)})
+		_, _ = kv.Put(ctx, key, passed)
+	}()
+	step, err := k.Passed(k.Wait(ctx, a))
+	if err != nil || step.Node != "ace" {
+		t.Fatalf("step %+v, %v", step, err)
+	}
+}

go/delivery/buildrun/asks_test.go

@@ -0,0 +1,43 @@
+package buildrun
+
+import (
+	"context"
+	"strings"
+	"testing"
+	"time"
+
+	"git.novox.be/novox/mesh-sdk/go/delivery"
+)
+
+func goAsk(id string) delivery.StepAsk {
+	return delivery.StepAsk{ID: id, Kind: delivery.AskDelivery, Language: delivery.AskGo, Asker: "mesh-delivery-builder", At: time.Now(),
+		Build: &delivery.BuildJob{Module: "m", Path: ".", Fetch: map[string]any{"repository": "novox/x", "commit": strings.Repeat("a", 40)},
+			Artifacts: []delivery.AskArtifact{{Artifact: "program"}}}}
+}
+
+// 1887/A4a: an asker puts the ask once and answers the step a build node wrote for it, naming the node; one that did not
+// pass is an error naming the node and why; a wait of an ask already put puts nothing.
+func TestAnAskerPutsOnceAndReadsTheStep(t *testing.T) {
+	f := &FakeAsks{Answer: func(a delivery.StepAsk) delivery.Step {
+		return delivery.Step{ID: a.ID, Node: "ace", State: delivery.StepPassed, Result: []byte(`{"artifacts":[]}`)}
+	}}
+	k := Asker{Publisher: f, Waiter: f}
+	step, err := k.Ask(context.Background(), goAsk("mvl0frv18d1957"))
+	if err != nil || step.Node != "ace" || len(f.Asks()) != 1 {
+		t.Fatalf("step %+v, %v, asks %d", step, err, len(f.Asks()))
+	}
+	f.Answer = func(a delivery.StepAsk) delivery.Step {
+		return delivery.Step{ID: a.ID, Node: "shanks", State: delivery.StepRefused, Reason: "go 1.27 installed"}
+	}
+	if _, err := k.Ask(context.Background(), goAsk("mvl0frv18d1958")); err == nil || !strings.Contains(err.Error(), "shanks refused: go 1.27 installed") {
+		t.Fatalf("a refused step: %v", err)
+	}
+	g := &FakeAsks{}
+	k = Asker{Publisher: g, Waiter: g, Bound: 20 * time.Millisecond}
+	if _, err := k.Wait(context.Background(), goAsk("mvl0frv18d1959")); err == nil || !strings.Contains(err.Error(), "did not end within") {
+		t.Fatalf("an ask whose step never ends: %v", err)
+	}
+	if len(g.Asks()) != 0 {
+		t.Fatal("a wait put an ask")
+	}
+}

go/delivery/stepasks_test.go

@@ -124,3 +124,24 @@ func TestAskSubjects(t *testing.T) {
 		t.Error("BuildSeat(agent)")
 	}
 }
+
+// 1887/A4b: a building key names the asks put for it; a done key or a name that is no ask id is refused.
+func TestABuildingKeyNamesItsAsks(t *testing.T) {
+	b := BuilderBuild{State: BuildBuilding, By: "novox.builder", Asks: []string{"mh3k2x9qa1b2c3"}}
+	raw, err := EncodeBuilderBuild(b)
+	if err != nil {
+		t.Fatal(err)
+	}
+	got, err := DecodeBuilderBuild(raw)
+	if err != nil || len(got.Asks) != 1 || got.Asks[0] != "mh3k2x9qa1b2c3" {
+		t.Fatalf("decoded %+v, %v", got, err)
+	}
+	for _, bad := range []BuilderBuild{
+		{State: BuildBuilding, Asks: []string{"postgres"}},
+		{State: BuildFailed, Error: "x", Asks: []string{"mh3k2x9qa1b2c3"}},
+	} {
+		if _, err := EncodeBuilderBuild(bad); err == nil {
+			t.Errorf("encoded %+v", bad)
+		}
+	}
+}