§535, §647, A-77, A-104: an ask no build seat holder takes within five minutes ends its step

The runner arms, before it puts an ask on a build seat's work queue, a key of its new state bounds (key-ttl) for five
minutes, deletes it when the holder's ask-taken is heard, and ends the step could not run naming the seat when the bus's
marker of the key's expiry is heard: no timer, and no look at the seat list.

Builds: design 47 G §32, §535, §602, §647, A-77, A-104

Claude-Session: https://claude.ai/code/session_01NcbgfGKaxw984XyAPfLm4h

modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline/main.go

@@ -53,6 +53,11 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
 	if err != nil {
 		return err
 	}
+	// The bound of each ask a build seat's holder must take: a key with its own time to live, its expiry heard (§535, A-77).
+	bounds, err := process.EssentialBucket(ctx, js, "pipeline runner", ownBucket(cred.Module, "bounds"), "bound an ask no holder takes")
+	if err != nil {
+		return err
+	}
 	// The registrations and each watched branch's catalog half, first views then events (§21, §195, §200): until a
 	// branch's view is held the runner acts on nothing new of it, and says so.
 	calls, err := process.Connect(cred, nats.Name(cred.Node+"."+cred.Module+".calls"), nats.RetryOnFailedConnect(true))
@@ -87,7 +92,8 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
 	// Its asking half (§22, §24): a pipeline's steps are placed from the module's registration and each build seat is asked
 	// for its step.
 	starter := &pipeline.Starter{Pipelines: createdKV{ctx: ctx, kv: pipelines}, Logf: logger.Printf, Say: say,
-		Registration: rv.Registrations.Registration, Steps: busStepAsker{ctx: ctx, js: js, cred: cred}, Name: cred.Module}
+		Registration: rv.Registrations.Registration, Steps: busStepAsker{ctx: ctx, js: js, cred: cred}, Name: cred.Module,
+		Arm: armBound(ctx, bounds)}
 	resume := func(branch string) {
 		half, ok := rv.Catalog.View(branch)
 		if !ok {
@@ -104,6 +110,12 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
 	if err := hearStepEnded(hub, hearer, logger.Printf); err != nil {
 		return err
 	}
+	if err := hearAskTaken(ctx, hub, bounds, logger.Printf); err != nil {
+		return err
+	}
+	if err := watchBounds(ctx, bounds, hearer, logger.Printf); err != nil {
+		return err
+	}
 	if err := hearDecided(hub, hearer, cred.Module, logger.Printf); err != nil {
 		return err
 	}

modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline/steps.go

@@ -10,6 +10,7 @@ import (
 	"github.com/nats-io/nats.go/jetstream"
 
 	seat "git.novox.be/novox/mesh-delivery/contracts/go/seats/meshpipelinerunner"
+	nodebuildgo "git.novox.be/novox/mesh-delivery/contracts/go/seats/nodebuildgo"
 	"git.novox.be/novox/mesh-sdk/go/catalogue"
 	"git.novox.be/novox/mesh-sdk/go/keys"
 	"git.novox.be/novox/mesh-sdk/go/process"
@@ -132,3 +133,69 @@ func hearStepEnded(hub *process.EventHub, h *pipeline.Hearer, logf func(string,
 	}
 	return nil
 }
+
+// hearAskTaken deletes the bound of each ask a build seat's holder took (§602, §647): the ask is answered, so its five minutes
+// end with no finding.
+func hearAskTaken(ctx context.Context, hub *process.EventHub, bounds jetstream.KeyValue, logf func(string, ...any)) error {
+	for _, ts := range pipeline.TakenSeats {
+		ts := ts
+		err := hub.Handle(ts.Subject, func(m jetstream.Msg) {
+			var ev nodebuildgo.AskTaken
+			if err := json.Unmarshal(m.Data(), &ev); err != nil || ev.Ask == "" {
+				logf("%s is no ask-taken: %v", m.Subject(), err)
+				_ = m.Term()
+				return
+			}
+			if err := bounds.Delete(ctx, ev.Ask); err != nil && !errors.Is(err, jetstream.ErrKeyNotFound) {
+				logf("the bound of ask %s, taken on %s, is not deleted, heard again: %v", ev.Ask, ev.Node, err)
+				_ = m.Nak()
+				return
+			}
+			_ = m.Ack()
+		})
+		if err != nil {
+			return err
+		}
+	}
+	return nil
+}
+
+// armBound is the bound of an ask: a key of the state bounds with its own time to live, named for the ask.
+func armBound(ctx context.Context, bounds jetstream.KeyValue) pipeline.Arm {
+	return func(ask string) error {
+		_, err := bounds.Create(ctx, ask, []byte(`{"ask":"`+ask+`"}`), jetstream.KeyTTL(pipeline.AskBound))
+		if errors.Is(err, jetstream.ErrKeyExists) {
+			return nil
+		}
+		return err
+	}
+}
+
+// watchBounds hears the end of each ask's bound, the bus's marker of the key's expiry, and ends the step no holder took (§32,
+// §535, A-77): no timer in the process.
+func watchBounds(ctx context.Context, bounds jetstream.KeyValue, h *pipeline.Hearer, logf func(string, ...any)) error {
+	w, err := bounds.WatchAll(ctx, jetstream.UpdatesOnly())
+	if err != nil {
+		return err
+	}
+	go func() {
+		for {
+			select {
+			case <-ctx.Done():
+				_ = w.Stop()
+				return
+			case e, ok := <-w.Updates():
+				if !ok {
+					return
+				}
+				if e == nil || e.Operation() != jetstream.KeyValuePurge || !pipeline.IsAskKey(e.Key()) {
+					continue
+				}
+				if err := h.NotTaken(e.Key(), pipeline.AskBound); err != nil {
+					logf("the step of ask %s, which no holder took, is not ended: %v", e.Key(), err)
+				}
+			}
+		}
+	}()
+	return nil
+}

modules/mesh-delivery-pipeline/internal/pipeline/bounds.go

@@ -0,0 +1,95 @@
+package pipeline
+
+import (
+	"errors"
+	"fmt"
+	"strings"
+	"time"
+
+	seat "git.novox.be/novox/mesh-delivery/contracts/go/seats/meshpipelinerunner"
+	nodebuildgo "git.novox.be/novox/mesh-delivery/contracts/go/seats/nodebuildgo"
+	nodebuildimage "git.novox.be/novox/mesh-delivery/contracts/go/seats/nodebuildimage"
+	nodebuildnode "git.novox.be/novox/mesh-delivery/contracts/go/seats/nodebuildnode"
+)
+
+// AskBound is how long an ask on a build seat's work queue may wait for its holder to take it (design 47 G §535, §647): five
+// minutes, the bound of a key of the state `bounds` with its own time to live (A-77, A-104), never a timer.
+const AskBound = 5 * time.Minute
+
+// TakenSeat is a build seat whose holder says `ask-taken` when it takes an ask (design 47 G §602): where the runner hears it.
+type TakenSeat struct {
+	Seat string
+	// Subject hears every node's ask-taken of the seat.
+	Subject string
+}
+
+// TakenSeats are the three build seats, from their contracts: the seat mesh-builds and the seat mesh-wants say no ask-taken,
+// so no bound is armed for their asks.
+var TakenSeats = []TakenSeat{
+	{Seat: nodebuildgo.Seat, Subject: nodebuildgo.SubjectAskTakenAll},
+	{Seat: nodebuildnode.Seat, Subject: nodebuildnode.SubjectAskTakenAll},
+	{Seat: nodebuildimage.Seat, Subject: nodebuildimage.SubjectAskTakenAll},
+}
+
+// BoundSeat says whether the holder of a seat says ask-taken, so an ask put on its work queue has a bound.
+func BoundSeat(name string) bool {
+	for _, s := range TakenSeats {
+		if s.Seat == name {
+			return true
+		}
+	}
+	return false
+}
+
+// Arm puts a bound of one ask's wait for its holder: a key of the state `bounds` named for the ask, written with its own
+// time to live. It is armed before the ask is put, so an ask-taken heard at once finds it.
+type Arm func(ask string) error
+
+// NotTaken ends the step asked under ask that no holder took within its bound, the bound's expiry heard (design 47 G §32,
+// §535, §647, A-77): the step ends `could not run` naming the seat, so its pipeline ends failed once the others end, never
+// a silent wait. A step that ended or was never asked of a seat takes nothing.
+func (h *Hearer) NotTaken(ask string, bound time.Duration) error {
+	pkeys, err := h.Store.Keys("*.*.*")
+	if err != nil {
+		return fmt.Errorf("the pipelines are not listed: %w", err)
+	}
+	for _, key := range pkeys {
+		p, _, err := h.Store.Get(key)
+		if err != nil || p.Outcome != seat.PipelineOutcomeRunning || stepOfAsk(p, ask) < 0 {
+			continue
+		}
+		return h.endNotTaken(key, ask, bound)
+	}
+	return nil
+}
+
+func (h *Hearer) endNotTaken(key, ask string, bound time.Duration) error {
+	for {
+		p, rev, err := h.Store.Get(key)
+		if err != nil {
+			return fmt.Errorf("the pipeline %s is not read: %w", key, err)
+		}
+		i := stepOfAsk(p, ask)
+		if i < 0 || p.Steps[i].Outcome != seat.StepResultOutcomeRunning {
+			return nil
+		}
+		now := h.now()
+		s := &p.Steps[i]
+		s.Outcome, s.Ended = seat.StepResultOutcomeCouldNotRun, &now
+		s.Why = fmt.Sprintf("the seat %s has no holder that took ask %s within %s", s.Seat, ask, bound)
+		ended := endPipeline(&p, now)
+		if _, err := h.Store.Update(key, p, rev); errors.Is(err, ErrChanged) {
+			continue
+		} else if err != nil {
+			return fmt.Errorf("pipeline %s cannot be written: %w", key, err)
+		}
+		h.logEnd(key, *s, now)
+		if err := h.say(seat.EventStepEnded, p, s); err != nil {
+			return fmt.Errorf("pipeline %s's %s is written but not said: %w", key, seat.EventStepEnded, err)
+		}
+		return h.said(p, false, ended)
+	}
+}
+
+// IsAskKey says whether a key of the state `bounds` is an ask's bound.
+func IsAskKey(key string) bool { return strings.HasPrefix(key, "step-") && !strings.Contains(key, ".") }

modules/mesh-delivery-pipeline/internal/pipeline/place.go

@@ -136,11 +136,20 @@ func stepAsk(p seat.Pipeline, e catalog.CatalogEntry, a catalog.Artifact, step,
 }
 
 // askAll puts each ask on its seat's work queue; the first refusal is named, the rest are still put.
-func askAll(asker StepAsker, key string, asks []Asked) error {
+func askAll(asker StepAsker, arm Arm, key string, asks []Asked) error {
 	var first error
 	for _, a := range asks {
 		accept, ok := acceptStep[a.Seat]
 		err := fmt.Errorf("the seat %s is no build seat the runner asks", a.Seat)
+		if ok {
+			// A build seat's holder says ask-taken: the ask's bound is armed before it is put (§535, §647).
+			if arm != nil && BoundSeat(a.Seat) {
+				if aerr := arm(a.ID); aerr != nil {
+					err = fmt.Errorf("its bound is not armed: %w", aerr)
+					ok = false
+				}
+			}
+		}
 		if ok {
 			err = asker.AskStep(accept, a.ID, a.Ask)
 		}

modules/mesh-delivery-pipeline/internal/pipeline/start.go

@@ -60,6 +60,8 @@ type Starter struct {
 	// asker.
 	Steps StepAsker
 	Name  string
+	// Arm arms the bound of an ask put on a build seat's work queue (§535, §647); nil arms none.
+	Arm Arm
 }
 
 // Start writes b's pipeline, running, and says pipeline-started; false, nil when its key was written already (the second
@@ -104,7 +106,7 @@ func (s *Starter) Start(b Begin) (bool, error) {
 		s.logf("pipeline-started of %s is not said: %v", key, err)
 	}
 	if s.Steps != nil && len(asks) > 0 {
-		if err := askAll(s.Steps, key, asks); err != nil {
+		if err := askAll(s.Steps, s.Arm, key, asks); err != nil {
 			return true, err
 		}
 	}
@@ -161,7 +163,7 @@ func (s *Starter) Resume(store Updater, branch string, half Half) (int, error) {
 		if len(due) == 0 {
 			continue
 		}
-		if err := askAll(s.Steps, key, due); err != nil && first == nil {
+		if err := askAll(s.Steps, s.Arm, key, due); err != nil && first == nil {
 			first = err
 		}
 		asked += len(due)

modules/mesh-delivery-pipeline/internal/pipeline/steps_test.go

@@ -9,6 +9,7 @@ import (
 	"git.novox.be/novox/mesh-contracts/operator-and-conversation/go/clients/operatorchannelclient"
 	oc "git.novox.be/novox/mesh-contracts/operator-and-conversation/go/seats/operatorchannel"
 	seat "git.novox.be/novox/mesh-delivery/contracts/go/seats/meshpipelinerunner"
+	wants "git.novox.be/novox/mesh-delivery/contracts/go/seats/meshwants"
 	nodebuildgo "git.novox.be/novox/mesh-delivery/contracts/go/seats/nodebuildgo"
 	"git.novox.be/novox/mesh-sdk/go/keys"
 )
@@ -298,3 +299,50 @@ func TestAnOptionalStepStillWaitingHoldsNoEnd(t *testing.T) {
 		}
 	}
 }
+
+// An ask no holder took within its bound ends its step `could not run` naming the seat, and with the rest ended the pipeline
+// failed (design 47 G §32, §535, §647, A-104); a step that ended, or ask nobody placed, ends nothing.
+func TestAnAskNoHolderTookEndsItsStepNamingTheSeat(t *testing.T) {
+	h, store, said, _, _, _ := fixture()
+	e := store[pkey]
+	e.p.Steps[1].Outcome = seat.StepResultOutcomePassed // the unit step ended; only lint is waiting for a holder
+	e.p.Steps[2].Outcome = seat.StepResultOutcomeNotAsked
+	store[pkey] = e
+	if err := h.NotTaken("nobody-asked", AskBound); err != nil || store[pkey].p.Steps[0].Outcome != seat.StepResultOutcomeRunning {
+		t.Fatalf("an ask no step names ended something: %v", err)
+	}
+	if err := h.NotTaken("a1", AskBound); err != nil {
+		t.Fatal(err)
+	}
+	p := store[pkey].p
+	if p.Steps[0].Outcome != seat.StepResultOutcomeCouldNotRun || !strings.Contains(p.Steps[0].Why, nodebuildgo.Seat) ||
+		p.Outcome != seat.PipelineOutcomeFailed || !strings.Contains(strings.Join(*said, " "), seat.EventPipelineEnded) {
+		t.Errorf("step %+v, pipeline %s (%s), said %v: want could not run naming %s and a failed pipeline", p.Steps[0], p.Outcome, p.Why, *said, nodebuildgo.Seat)
+	}
+	// Its end taken once: the bound's expiry heard again changes nothing.
+	n := len(*said)
+	if err := h.NotTaken("a1", AskBound); err != nil || len(*said) != n {
+		t.Errorf("a second expiry said %d more events: %v", len(*said)-n, err)
+	}
+}
+
+// The bound is armed before the ask is put, only for the build seats whose holder says ask-taken, and an ask whose bound is
+// not armed is not put (§535, §647).
+func TestTheBoundIsArmedBeforeAnAskIsPut(t *testing.T) {
+	var order []string
+	asker := askerFunc(func(accept, id string, _ any) error { order = append(order, "put "+id); return nil })
+	arm := Arm(func(ask string) error { order = append(order, "arm "+ask); return nil })
+	asks := []Asked{{Seat: nodebuildgo.Seat, ID: "step-1", Ask: 1}, {Seat: wants.Seat, ID: "step-2", Ask: 2}}
+	if err := askAll(asker, arm, "k", asks); err != nil || strings.Join(order, ",") != "arm step-1,put step-1,put step-2" {
+		t.Errorf("order %v, %v; want the build seat's bound armed first and mesh-wants' ask with none", order, err)
+	}
+	order = nil
+	failing := Arm(func(string) error { return errors.New("bucket gone") })
+	if err := askAll(asker, failing, "k", asks[:1]); err == nil || len(order) != 0 {
+		t.Errorf("an ask whose bound cannot be armed was put or said nothing: %v %v", order, err)
+	}
+}
+
+type askerFunc func(accept, id string, body any) error
+
+func (f askerFunc) AskStep(accept, id string, body any) error { return f(accept, id, body) }

modules/mesh-delivery-pipeline/manifest_test.go

@@ -21,8 +21,10 @@ func TestManifest(t *testing.T) {
 		t.Fatalf("the manifest does not read: %v", err)
 	}
 	var states, streams, objects []string
+	keyTTL := map[string]bool{}
 	for _, s := range m.State {
 		states = append(states, s.Name)
+		keyTTL[s.Name] = s.KeyTTL
 	}
 	for _, s := range m.Streams {
 		streams = append(streams, s.Name)
@@ -48,7 +50,9 @@ func TestManifest(t *testing.T) {
 		{"claims the seat serving every verb of its definition", len(m.Claims) == 1 && m.Claims[0].Name == seat.Seat &&
 			equalSets(m.Claims[0].Serves, seat.Verbs())},
 		// §27, §85, §330, §355, §77: the old state sections.
-		{"old state sections", equalSets(states, []string{seat.StatePipelines, seat.StateKept, "aged", "findings", "first-views"})},
+		{"old state sections", equalSets(states, []string{seat.StatePipelines, seat.StateKept, "aged", "bounds", "findings", "first-views"})},
+		// §535, §647, A-77, A-104: an ask no holder takes is bound by a key with its own time to live.
+		{"bounds is key-ttl", keyTTL["bounds"]},
 		// §27, §85: the step log on an old stream, its outputs in an old objects section.
 		// RN24 (§383, §387): beside it, the finding stream its findings are said on.
 		{"old streams section", equalSets(streams, []string{seat.StateStepLog, "mesh-delivery-pipeline-findings"})},
@@ -63,7 +67,7 @@ func TestManifest(t *testing.T) {
 		{"consumes", equalSets(m.Consumes, []string{
 			"mesh-delivery-indexer.take-in.taken", "mesh-builds.package-published", "mesh-builds.toolchain-agreed", "mesh-builds.step-ended",
 			"operator-channel.decided", "node-build-go.step-ended", "node-build-node.step-ended",
-			"node-build-image.step-ended", "mesh-wants.step-ended", "mesh-catalog.registration-changed",
+			"node-build-image.step-ended", "node-build-go.ask-taken", "node-build-node.ask-taken", "node-build-image.ask-taken", "mesh-wants.step-ended", "mesh-catalog.registration-changed",
 			"mesh-catalog.catalog-revision", "mesh-conditions.findings-asked"})},
 		// RN2 (§21, §195, §200, §77): the registrations and the catalog half asked by their verbs, the first views read and
 		// asked again by hand (a hand act).

modules/mesh-delivery-pipeline/module.json

@@ -35,6 +35,9 @@
     "node-build-go.step-ended",
     "node-build-node.step-ended",
     "node-build-image.step-ended",
+    "node-build-go.ask-taken",
+    "node-build-node.ask-taken",
+    "node-build-image.ask-taken",
     "mesh-wants.step-ended",
     "mesh-catalog.registration-changed",
     "mesh-catalog.catalog-revision",
@@ -68,6 +71,11 @@
       "name": "aged",
       "history": 1
     },
+    {
+      "name": "bounds",
+      "history": 1,
+      "key-ttl": true
+    },
     {
       "name": "findings",
       "history": 1