Merge remote-tracking branch 'origin/feat/r4g' into feat/r4g

# Conflicts:
#	modules/mesh-delivery-builder/internal/build/run_test.go

modules/mesh-delivery-builder/cmd/mesh-delivery-builder/held.go

@@ -0,0 +1,74 @@
+package main
+
+import (
+	"context"
+	"errors"
+	"fmt"
+
+	"github.com/nats-io/nats.go/jetstream"
+
+	"git.novox.be/novox/mesh-sdk/go/delivery"
+)
+
+// heldOnTheBus says whether an ask a build key names is on its build seat's stream, or a build node wrote a step for it
+// (issue 1910): read once, at the take-over of a key at start, never polled.
+type heldOnTheBus struct{ js jetstream.JetStream }
+
+// Held is build.Builder's Held: a step key `<node>.<ask id>` in the build module's steps, or a message of the stream under
+// the ask's subject whose Nats-Msg-Id is the ask's id.
+func (h heldOnTheBus) Held(ctx context.Context, a delivery.StepAsk) (bool, error) {
+	module, err := delivery.BuildModuleOf(a.Language)
+	if err != nil {
+		return false, err
+	}
+	pattern, err := delivery.AskStepWatch(a.ID)
+	if err != nil {
+		return false, err
+	}
+	steps, err := h.js.KeyValue(ctx, delivery.StepsBucket(module))
+	if err != nil {
+		return false, fmt.Errorf("the bucket %s: %w", delivery.StepsBucket(module), err)
+	}
+	keys, err := steps.ListKeysFiltered(ctx, pattern)
+	if err != nil {
+		return false, err
+	}
+	for range keys.Keys() {
+		keys.Stop()
+		return true, nil // a node took it
+	}
+	subject, err := delivery.AskSubject(a.Language, a.Kind)
+	if err != nil {
+		return false, err
+	}
+	name, err := h.js.StreamNameBySubject(ctx, subject)
+	if errors.Is(err, jetstream.ErrStreamNotFound) {
+		return false, nil
+	}
+	if err != nil {
+		return false, err
+	}
+	stream, err := h.js.Stream(ctx, name)
+	if err != nil {
+		return false, err
+	}
+	// A seat's stream is a work queue, which takes no second consumer over its subjects: its messages are read by sequence
+	// instead, each a direct read that consumes nothing.
+	info, err := stream.Info(ctx)
+	if err != nil {
+		return false, err
+	}
+	for seq := info.State.FirstSeq; seq != 0 && seq <= info.State.LastSeq; seq++ {
+		m, err := stream.GetMsg(ctx, seq)
+		if errors.Is(err, jetstream.ErrMsgNotFound) {
+			continue // taken and acknowledged, or removed
+		}
+		if err != nil {
+			return false, err
+		}
+		if m.Subject == subject && m.Header.Get(jetstream.MsgIDHeader) == a.ID {
+			return true, nil
+		}
+	}
+	return false, nil
+}

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

@@ -156,6 +156,7 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
 	b := &build.Builder{
 		Claimer:    build.Claimer{Store: kvStore{builds}, By: cred.Node + "." + cred.Module},
 		Asks:       buildrun.Asker{Publisher: asks, Waiter: asks, Bound: bound},
+		Held:       heldOnTheBus{js},
 		Toolchains: tools,
 		Manifests:  manifestReader{manifests},
 		Logf:       logger.Printf,

modules/mesh-delivery-builder/internal/build/run.go

@@ -35,6 +35,12 @@ type Builder struct {
 		Put(ctx context.Context, a delivery.StepAsk) error
 		Wait(ctx context.Context, a delivery.StepAsk) (delivery.Step, error)
 	}
+	// Held says whether an ask a key names is on its stream or a build node wrote a step for it (issue 1910): a key taken
+	// over at start whose ask is neither was left by a builder that stopped between naming the ask and putting it, and
+	// the ask is put then, under the same id. Nil takes every named ask as held.
+	Held interface {
+		Held(ctx context.Context, a delivery.StepAsk) (bool, error)
+	}
 	Toolchains ToolchainEntries
 	Manifests  Manifests
 	// Limit is how many builds the builder has asked at once; their asks wait in the build seats' streams, which are the
@@ -668,6 +674,21 @@ func (b *Builder) runLane(ctx context.Context, l *lane, repository string, m del
 	ask := delivery.StepAsk{ID: buildstep.NewID(), Kind: kind, Language: language, Asker: delivery.BuilderModule, At: time.Now(), Build: job}
 	if id := asked.resume(language); id != "" {
 		ask.ID = id // put before this builder started again: the stream holds it, or a node already ran it
+		if b.Held != nil {
+			held, err := b.Held.Held(ctx, ask)
+			if err != nil {
+				l.err = fmt.Errorf("whether the ask %s of %s the key names is on its stream cannot be read: %w", ask.ID, l.module, err)
+				return
+			}
+			if !held {
+				// The builder that named it stopped before it put it (issue 1910): put it now, under the same id.
+				b.logf("the ask %s of %s that the key names was never put: it is put now", ask.ID, l.module)
+				if err := b.Asks.Put(ctx, ask); err != nil {
+					l.err = err
+					return
+				}
+			}
+		}
 	} else {
 		if err := asked.note(ctx, language, ask.ID, "asked "+l.module); err != nil {
 			l.err = fmt.Errorf("the ask of %s could not be written to the key: %w", l.module, err)

modules/mesh-delivery-builder/internal/build/run_test.go

@@ -636,3 +636,38 @@ func TestABuildAskWaitingInItsStreamIsNotBounded(t *testing.T) {
 		t.Fatalf("the key = %+v after %d asks, want done from the one ask", v, len(r.asks.Asks()))
 	}
 }
+
+// heldAsks says an ask is held when its id is among held.
+type heldAsks map[string]bool
+
+func (h heldAsks) Held(_ context.Context, a delivery.StepAsk) (bool, error) { return h[a.ID], nil }
+
+// 1910: a builder that stopped between naming the ask in its key and putting it left a key naming an ask neither the
+// stream holds nor any node ran: started again, the builder puts that ask, under the same id, once, and the key is
+// built from its step; an ask the stream holds is not put again.
+func TestAnAskTheKeyNamesAndNobodyPutIsPutAtStart(t *testing.T) {
+	for _, held := range []bool{false, true} {
+		r := newRig(t, goBundle)
+		const id = "mvl0frv18d1958"
+		data, _ := delivery.EncodeBuilderBuild(delivery.BuilderBuild{State: delivery.BuildBuilding, By: "novox.builder", Manifest: manifestD,
+			Step: "asked mesh-build-go", Asks: map[string]string{delivery.AskGo: id}})
+		r.st.vals[r.key] = data
+		r.b.Held = heldAsks{id: held}
+		if held {
+			r.asks.Answer = nil
+			r.asks.End(delivery.Step{ID: id, Node: "shanks", State: delivery.StepPassed,
+				Result: []byte(`{"artifacts":[{"artifact":"tool","digest":"` + digest + `","reference":"artifact-store://m/tool/blobs/` + digest + `"}]}`)})
+		}
+		v := r.consider(t, "modules/m")
+		asks := r.asks.Asks()
+		switch {
+		case held && len(asks) != 0:
+			t.Errorf("an ask the stream holds was put again: %+v", asks)
+		case !held && (len(asks) != 1 || asks[0].ID != id):
+			t.Errorf("the ask nobody put: asks %+v, want one under %s", asks, id)
+		}
+		if v.State != delivery.BuildDone {
+			t.Errorf("held %v: the key = %+v, want done", held, v)
+		}
+	}
+}

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

@@ -197,7 +197,6 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
 	}
 	tools := &buildrun.ToolchainSet{}
 	presence := &buildrun.PresenceSet{}
-	have := &buildrun.HaveSet{}
 	trunks := &check.Trunks{}
 	// A check is one ask per pull request and build module on the build seats' check stream, pulled by a build node with a
 	// free slot (novox/hq ADR 0525 §1, §3): the checker chooses no node and starts no step.
@@ -279,8 +278,10 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
 		defer cc.Stop()
 	}
 	errs := make(chan error, 8)
-	// The watchman's presence says which nodes are asleep. Where its bucket is not there yet no node is online, so nothing
-	// is checked; the checker says so and is started again with the watchman.
+	// The watchman's presence: a node coming back online asks again a check whose forge was not heard (UnheardWords), and
+	// the assigner's own presence says whether a composition is being made (compositionWatch, ADR 0537 §4). A check is no
+	// longer held for a node: its asks wait in the build seats' streams, which redeliver a lost node's ask (ADR 0525 §6,
+	// issue 1912), so the node-engine's have is not watched here.
 	switch kv, err := js.KeyValue(ctx, delivery.PresenceBucket); {
 	case errors.Is(err, jetstream.ErrBucketNotFound):
 		logger.Printf("the bucket %s does not exist yet: no node counts as online until the checker starts again", delivery.PresenceBucket)
@@ -294,27 +295,7 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
 		go func() {
 			errs <- follow(ctx, pw, func(e jetstream.KeyValueEntry) {
 				if presence.Put(e.Key(), e.Value(), !opOf(e)) {
-					ch.NodesChanged(ctx) // a node came online or went: a check that waited for one may go on
-				}
-			})
-		}()
-	}
-	// The node-engine's have says whether a node's build module runs and is healthy: a node whose build module is not is not
-	// chosen. Where its bucket is not there yet no build module counts as healthy, so nothing is checked.
-	switch kv, err := js.KeyValue(ctx, delivery.HaveBucket); {
-	case errors.Is(err, jetstream.ErrBucketNotFound):
-		logger.Printf("the bucket %s does not exist yet: no build module counts as healthy until the checker starts again", delivery.HaveBucket)
-	case err != nil:
-		return fmt.Errorf("the bucket %s: %w", delivery.HaveBucket, err)
-	default:
-		hw, err := settle(ctx, kv, func(e jetstream.KeyValueEntry) { have.Put(e.Key(), e.Value(), !opOf(e)) })
-		if err != nil {
-			return err
-		}
-		go func() {
-			errs <- follow(ctx, hw, func(e jetstream.KeyValueEntry) {
-				if have.Put(e.Key(), e.Value(), !opOf(e)) {
-					ch.NodesChanged(ctx)
+					ch.NodesChanged(ctx) // a node came online or went: a check whose forge was not heard may now be heard
 				}
 			})
 		}()

modules/mesh-delivery-checker/cmd/mesh-delivery-checker/nodes_test.go

@@ -0,0 +1,21 @@
+package main
+
+import (
+	"os"
+	"strings"
+	"testing"
+)
+
+// 1912: with checks as asks on the build seats' streams, the stream redelivers a lost node's ask (ADR 0525 §6), so the
+// checker program opens no node-engine have to start again a check that failed for want of a node.
+func TestTheCheckerWatchesNoHave(t *testing.T) {
+	raw, err := os.ReadFile("main.go")
+	if err != nil {
+		t.Fatal(err)
+	}
+	for _, s := range []string{"delivery.HaveBucket", "buildrun.HaveSet"} {
+		if strings.Contains(string(raw), s) {
+			t.Errorf("the checker program still reads %s", s)
+		}
+	}
+}

modules/mesh-delivery-checker/internal/check/checker.go

@@ -17,10 +17,6 @@ import (
 // MESH_CHECKER_BOUND still ends a check that ran past it.
 const DefaultBound time.Duration = 0
 
-// NodesWords is in the Words of a failure that is the build nodes' (none qualifies, the toolchains disagree): such a verdict
-// is judged again when the nodes change, by a watch and never a timer (ADR 0437 §8).
-const NodesWords = "build node qualifies"
-
 // UnheardWords end the Words of a failure because a part of the mesh the check reads was not heard (the forge, issue 1624):
 // like a failure of the nodes, it is not the change's, and is judged again when the nodes change.
 const UnheardWords = "not heard; judged again when the nodes change"
@@ -310,9 +306,11 @@ func (ch *Checker) claim(ctx context.Context, cs *Case, first bool, epoch uint64
 	return &claimed{rev}, nil
 }
 
-// IsNodesFailure says whether a verdict failed because no build node could take the check.
+// IsNodesFailure says whether a verdict failed because a part of the mesh the check reads was not heard (UnheardWords):
+// it is judged again when the nodes change. No check fails for want of a build node any more: its asks wait in the build
+// seats' streams (ADR 0525 §6, issue 1912).
 func IsNodesFailure(v delivery.Verdict) bool {
-	return v.State == delivery.VerdictFail && (strings.Contains(v.Failed.Words, NodesWords) || strings.Contains(v.Failed.Words, UnheardWords))
+	return v.State == delivery.VerdictFail && strings.Contains(v.Failed.Words, UnheardWords)
 }
 
 // try claims the check, judges it within its bound and writes the verdict and the owed statuses. A check that ends

modules/mesh-delivery-checker/internal/check/checker_test.go

@@ -208,11 +208,13 @@ func TestACheckPastItsBoundFailsNamingTheStuckStep(t *testing.T) { // N6/AC3
 	}
 }
 
-func TestNoQualifyingNodeFailsWithTheReasonAndIsJudgedAgainWhenTheNodesChange(t *testing.T) { // N6/AC4
+// A check whose forge was not heard is judged again when the nodes change (issue 1912: no check fails for want of a build
+// node any more; its asks wait in the streams).
+func TestAnUnheardForgeFailsWithTheReasonAndIsJudgedAgainWhenTheNodesChange(t *testing.T) { // N6/AC4
 	ctx := context.Background()
 	j := &judgeFunc{f: func(_ context.Context, _ *Case, n int) (Outcome, error) {
 		if n == 1 {
-			return Outcome{Verdict: failure("shanks", "no go "+NodesWords+": none is online")}, nil
+			return Outcome{Verdict: failure("the forge", "novox/mesh-delivery#7: go.mod cannot be read: no responders (the forge was "+UnheardWords+")")}, nil
 		}
 		return Outcome{Verdict: passing()}, nil
 	}}

modules/mesh-delivery-checker/internal/check/rank_test.go

@@ -39,7 +39,7 @@ func TestCheckerStatusNamesTheWaitingChecksByKind(t *testing.T) { // 1897/AC4
 	ctx := context.Background()
 	j := &judgeFunc{f: func(_ context.Context, c *Case, _ int) (Outcome, error) {
 		if c.Group == group(2) {
-			return Outcome{Verdict: failure("checker", "no "+NodesWords+": every node is offline")}, nil
+			return Outcome{Verdict: failure("the forge", "go.mod cannot be read: no responders (the forge was "+UnheardWords+")")}, nil
 		}
 		return Outcome{Verdict: passing()}, nil
 	}}

modules/mesh-delivery-view/manifest_test.go

@@ -94,9 +94,9 @@ func TestManifestIsTakenAndReadsOnly(t *testing.T) { // 1134/AC7
 			t.Errorf("the manifest declares %q: the view writes only its answers", k)
 		}
 	}
-	// The view reads the build seats' streams for the backlog of asks (issue 1887 A8a): the three seats it uses.
-	if want := []string{"node-build-go", "node-build-image", "node-build-node"}; !reflect.DeepEqual(m.Uses, want) {
-		t.Errorf("uses = %v, want %v", m.Uses, want)
+	// The view uses no build seat: it reads their workers by the broker's named grant and publishes no ask (ADR 0525 §2).
+	if len(m.Uses) != 0 {
+		t.Errorf("uses = %v, want none", m.Uses)
 	}
 	// The one state the view writes: the operator's answers to the asks that need no proof (ADR 0528 clause 6, 1297/B9), which the teller reads.
 	if want := []state{{Name: "answers", History: 5}}; !reflect.DeepEqual(m.State, want) {

modules/mesh-delivery-view/module.json

@@ -28,11 +28,6 @@
     "mesh-delivery-watchman.stuck",
     "node-engine.have"
   ],
-  "uses": [
-    "node-build-go",
-    "node-build-image",
-    "node-build-node"
-  ],
   "own-secrets": {
     "broker": "${dir:state}/broker"
   },

modules/mesh-delivery-watchman/cmd/mesh-delivery-watchman/integration_test.go

@@ -167,23 +167,23 @@ func TestWatchmanWhole(t *testing.T) { // 1135/AC8
 
 	// An ask put on a build seat gets a pulse; one no node takes lets the pulse expire, and that writes a stuck key.
 	askStream, _ := delivery.AskStream(delivery.AskGo)
-	askSubject, _ := delivery.AskSubject(delivery.AskGo, delivery.AskCheck)
-	putAsk := func(id string) {
+	steps := kvs[delivery.StepsBucket("mesh-build-go")]
+	// A node takes an ask: it writes the step key running (and passed when the ask ends).
+	putStep := func(id, state string) {
 		t.Helper()
-		body, err := delivery.EncodeStepAsk(delivery.StepAsk{ID: id, Kind: delivery.AskCheck, Language: delivery.AskGo, Asker: "mesh-delivery-checker", At: time.Now().UTC(),
-			Check: &delivery.CheckJob{Fetch: map[string]any{"repository": "novox/x", "commit": "0123456789abcdef0123456789abcdef01234567"}, Levels: []string{delivery.LevelUnit}}})
+		body, err := delivery.EncodeStep(delivery.Step{ID: id, Tool: "build_go_test", Node: "n1", State: state, Started: time.Now().UTC()})
 		if err != nil {
 			t.Fatal(err)
 		}
-		if err := ctl.Publish(askSubject, body); err != nil {
+		if _, err := steps.Put(ctx, "n1."+id, body); err != nil {
 			t.Fatal(err)
 		}
 	}
 	const ask1, ask2 = "m0000001aaaaaa", "m0000002bbbbbb"
-	putAsk(ask1)
-	until("a pulse for the ask waiting", func() bool { _, ok := has(delivery.PulseBucket, pulseOf(askStream, ask1)); return ok })
+	putStep(ask1, delivery.StepRunning)
+	until("a pulse for the ask a node took", func() bool { _, ok := has(delivery.PulseBucket, pulseOf(askStream, ask1)); return ok })
 	var stuck delivery.Stuck
-	until("a stuck key for the ask no node took", func() bool {
+	until("a stuck key for the ask no node ended", func() bool {
 		e, ok := has(delivery.StuckBucket, stuckOf(askStream, ask1))
 		if !ok {
 			return false
@@ -210,11 +210,9 @@ func TestWatchmanWhole(t *testing.T) { // 1135/AC8
 	until("node n2 online again", func() bool { on, ok := online("n2"); return ok && on })
 	until("the pulse of node n2 deleted", func() bool { _, ok := has(delivery.PulseBucket, pulseOf(delivery.PresenceBucket, "n2")); return !ok })
 	// An ask a node takes at once is the same: no stuck key, however long the bound passes.
-	putAsk(ask2)
+	putStep(ask2, delivery.StepRunning)
 	until("a pulse for the second ask", func() bool { _, ok := has(delivery.PulseBucket, pulseOf(askStream, ask2)); return ok })
-	if _, err := kvs[delivery.StepsBucket("mesh-build-go")].Put(ctx, "n1."+ask2, []byte(`{}`)); err != nil {
-		t.Fatal(err)
-	}
+	putStep(ask2, delivery.StepPassed)
 	until("the pulse of the second ask deleted", func() bool { _, ok := has(delivery.PulseBucket, pulseOf(askStream, ask2)); return !ok })
 	time.Sleep(6 * time.Second) // twice the bound and the marker time
 	for _, k := range []string{stuckOf(delivery.PresenceBucket, "n2"), stuckOf(askStream, ask2)} {
@@ -223,3 +221,84 @@ func TestWatchmanWhole(t *testing.T) { // 1135/AC8
 		}
 	}
 }
+
+// TestAnAskRunningBeforeTheStartGetsItsPulse puts a running step in the build module's steps before the watchman starts, and
+// sees the pulse of its ask once it is up: what stood at the start is mirrored as what happens later is (issue 1909).
+func TestAnAskRunningBeforeTheStartGetsItsPulse(t *testing.T) { // 1909
+	dir, err := os.MkdirTemp(os.Getenv("TMPDIR"), "w")
+	if err != nil {
+		t.Fatal(err)
+	}
+	t.Cleanup(func() { os.RemoveAll(dir) })
+	conf := filepath.Join(dir, "bus.conf")
+	if err := os.WriteFile(conf, []byte(fmt.Sprintf(busConf, filepath.Join(dir, "js"))), 0o600); err != nil {
+		t.Fatal(err)
+	}
+	opts, err := server.ProcessConfigFile(conf)
+	if err != nil {
+		t.Fatal(err)
+	}
+	opts.NoLog, opts.NoSigs = true, true
+	srv, err := server.NewServer(opts)
+	if err != nil {
+		t.Fatal(err)
+	}
+	go srv.Start()
+	if !srv.ReadyForConnections(10 * time.Second) {
+		t.Fatal("the bus did not come up")
+	}
+	t.Cleanup(srv.Shutdown)
+	ctl, err := nats.Connect(srv.ClientURL(), nats.UserInfo("ctl", "pw"))
+	if err != nil {
+		t.Fatal(err)
+	}
+	t.Cleanup(ctl.Close)
+	js, err := jetstream.New(ctl)
+	if err != nil {
+		t.Fatal(err)
+	}
+	ctx, cancel := context.WithTimeout(context.Background(), 60*time.Second)
+	defer cancel()
+	kvs := map[string]jetstream.KeyValue{}
+	for _, c := range []jetstream.KeyValueConfig{
+		{Bucket: delivery.PresenceBucket, History: 20},
+		{Bucket: delivery.StuckBucket, History: 20},
+		{Bucket: delivery.PulseBucket, LimitMarkerTTL: time.Second},
+		{Bucket: delivery.StepsBucket("mesh-build-go"), History: 5},
+	} {
+		kv, err := js.CreateKeyValue(ctx, c)
+		if err != nil {
+			t.Fatal(err)
+		}
+		kvs[c.Bucket] = kv
+	}
+	const id = "m0000009ffffff"
+	body, err := delivery.EncodeStep(delivery.Step{ID: id, Tool: "build_go_test", Node: "n1", State: delivery.StepRunning, Started: time.Now().UTC()})
+	if err != nil {
+		t.Fatal(err)
+	}
+	if _, err := kvs[delivery.StepsBucket("mesh-build-go")].Put(ctx, "n1."+id, body); err != nil {
+		t.Fatal(err)
+	}
+	broker := filepath.Join(dir, "broker")
+	cred, _ := json.Marshal(credential{URL: srv.ClientURL(), Node: "n1", Module: "mesh-delivery-watchman", User: "n1.mesh-delivery-watchman", Password: "pw"})
+	if err := os.WriteFile(broker, cred, 0o600); err != nil {
+		t.Fatal(err)
+	}
+	env := map[string]string{"MESH_BROKER_FILE": broker}
+	runCtx, stop := context.WithCancel(ctx)
+	done := make(chan error, 1)
+	go func() { done <- run(runCtx, func(k string) string { return env[k] }, io.Discard) }()
+	t.Cleanup(func() { stop(); <-done })
+	stream, _ := delivery.AskStream(delivery.AskGo)
+	pk, err := delivery.PulseKey(stream, id)
+	if err != nil {
+		t.Fatal(err)
+	}
+	for deadline := time.Now().Add(20 * time.Second); time.Now().Before(deadline); time.Sleep(50 * time.Millisecond) {
+		if _, err := kvs[delivery.PulseBucket].Get(ctx, pk); err == nil {
+			return
+		}
+	}
+	t.Fatalf("no pulse for the ask running before the start (%s)", pk)
+}

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

@@ -1,5 +1,5 @@
 // Command mesh-delivery-watchman is delivery's watchman (novox/hq ADR 0429 §1, ADR 0431, ADR 0433, issue 1135): it
-// hears the bus's connect and disconnect events and writes presence, mirrors every move a machine owes (a build ask waiting for a node,
+// hears the bus's connect and disconnect events and writes presence, mirrors every move a machine owes (a build ask a node is running,
 // a node or part gone offline) into the pulse bucket with a time to live, and writes
 // a stuck key when the server expires one. It holds no timer: every reaction is an event, a watch or a request at
 // start or at a reconnect, and the server's expiry is the only clock.
@@ -179,15 +179,10 @@ func follow(ctx context.Context, w jetstream.KeyWatcher, each func(jetstream.Key
 	}
 }
 
-// stepTaken is a change of a build module's steps: a step key written for an ask is that ask taken by a node. The node's
-// `paused.<node>` key and a key that is no step key are left.
+// stepTaken is a change of a build module's steps, handed to the watchman as it is (watch.Step): a running step is its ask
+// taken by a node and pulsed, one that ended ends the pulse. An ask still waiting in a stream has no pulse (issue 1908).
 func stepTaken(w *watch.Watchman, language string, e jetstream.KeyValueEntry) {
-	if opOf(e) != watch.Put {
-		return
-	}
-	if _, id, err := delivery.ParseStepKey(e.Key()); err == nil {
-		w.Taken(language, id)
-	}
+	w.Step(language, e.Key(), e.Value(), opOf(e))
 }
 
 // run is the program: it returns when ctx ends or the bus cannot be used.
@@ -318,23 +313,6 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
 			}
 		}
 	}
-	// The asks put on the build seats' streams: a plain subscription hears each as it is published, as the stream captures it.
-	for _, lang := range delivery.AskLanguages {
-		for _, kind := range delivery.AskKinds {
-			subject, _ := delivery.AskSubject(lang, kind)
-			lang := lang
-			if _, err := conn.Subscribe(subject, func(m *nats.Msg) {
-				a, err := delivery.DecodeStepAsk(m.Data)
-				if err != nil {
-					logger.Printf("an ask on %s is not readable: %v", m.Subject, err)
-					return
-				}
-				w.Asked(lang, a.ID, a.At)
-			}); err != nil {
-				return err
-			}
-		}
-	}
 	for subject, open := range map[string]bool{connectSubject: true, disconnectSubject: false} {
 		if _, err := conn.Subscribe(subject, onEvent(open)); err != nil {
 			return err

modules/mesh-delivery-watchman/internal/watch/mirror.go

@@ -2,41 +2,53 @@ package watch
 
 import (
 	"git.novox.be/novox/mesh-sdk/go/delivery"
-	"time"
 )
 
-// A waiting ask is a move (novox/hq ADR 0525 §4, issue 1887 A8b): a message on a build seat's stream (mesh-sdk go/delivery
-// AskStream) that a node of the seat must pull. Its pulse is the pulse of the bucket named as the stream, under the ask's id;
-// the ask put on the seat is the move beginning at its time, and the ask taken (its step key `<node>.<ask id>` written in
-// the build module's steps, delivery.StepsBucket) is the move done. An ask no node takes within the bound is stuck, as any
-// move. The watchman reads no bucket of the controller for it.
+// A build ask taken by a node is a move (novox/hq ADR 0525 §4, issues 1887 A8b, 1908, 1909): the node writes the ask's step
+// key `<node>.<ask id>` running in the build module's steps (delivery.StepsBucket) when it takes the ask, and writes it passed,
+// failed or refused when the ask ends. The pulse of the move is the pulse of the bucket named as the seat's stream
+// (delivery.AskStream), under the ask's id, and it begins when the node started the step.
+//
+// **An ask still waiting in the stream is a backlog, not a stall** (issue 1908): the rebuild after the switch puts about 170
+// asks on a few nodes' slots, and most wait for hours with every worker healthy. No pulse is made for a waiting ask, so none
+// can expire into a false stuck row; the view shows the backlog and the oldest waiting ask's age (delivery view, A8a). What
+// is read of the steps at the watchman's start is mirrored the same way, so an ask running when the watchman was down is
+// pulsed again (issue 1909).
 
-// AskedKey names a waiting ask for the pulse and stuck keys: the seat's stream and the ask's id.
-func askedBucket(language string) (string, error) { return delivery.AskStream(language) }
-
-// Asked is an ask of a build seat of language put on its stream at time at (the ask's own time).
-func (w *Watchman) Asked(language, id string, at time.Time) {
-	bucket, err := askedBucket(language)
+// Step is a change of a key of the steps bucket of language's build module, read from the watch (at the start the keys that
+// stand, then each change). A running step makes the pulse of its ask, from the step's own start; a step that ended, or a
+// key deleted, ends it. The node's `paused.<node>` key, a key that is no step key and a value that cannot be read are left.
+func (w *Watchman) Step(language, key string, value []byte, op Op) {
+	bucket, err := delivery.AskStream(language)
 	if err != nil {
-		w.say("no pulse for the ask %s: %v", id, err)
+		w.say("no pulse for the step %s: %v", key, err)
 		return
 	}
-	w.mu.Lock()
-	defer w.mu.Unlock()
-	if at.IsZero() {
-		at = w.now()
-	}
-	w.mirror(bucket, id, describe("the ask %s to be taken by a node", id), "", at, 0)
-}
-
-// Taken is the ask taken by a node: its step key written. The pulse is deleted, so it never expires, and a stuck key it left is
-// written as moved.
-func (w *Watchman) Taken(language, id string) {
-	bucket, err := askedBucket(language)
+	node, id, err := delivery.ParseStepKey(key)
 	if err != nil {
 		return
 	}
 	w.mu.Lock()
 	defer w.mu.Unlock()
-	w.moved(bucket, id)
+	switch op {
+	case Expired:
+		return
+	case Delete:
+		w.moved(bucket, id)
+		return
+	}
+	step, err := delivery.DecodeStep(value)
+	if err != nil {
+		w.say("the step %s cannot be read, and is left: %v", key, err)
+		return
+	}
+	if step.State != delivery.StepRunning {
+		w.moved(bucket, id)
+		return
+	}
+	since := step.Started
+	if since.IsZero() {
+		since = w.now()
+	}
+	w.mirror(bucket, id, describe("the ask %s to end on %s", id, node), node, since, 0)
 }

modules/mesh-delivery-watchman/internal/watch/watch_test.go

@@ -78,6 +78,23 @@ const (
 	presenceNS = delivery.PresenceBucket
 )
 
+// runningStep is a build module's step key as a node writes it when it takes an ask: the ask's id, running, since started.
+func runningStep(id string, started time.Time) []byte {
+	b, err := delivery.EncodeStep(delivery.Step{ID: id, Tool: "build_go_test", Node: "n1", State: delivery.StepRunning, Started: started})
+	if err != nil {
+		panic(err)
+	}
+	return b
+}
+
+func endedStep(id, state string, started time.Time) []byte {
+	b, err := delivery.EncodeStep(delivery.Step{ID: id, Tool: "build_go_test", Node: "n1", State: state, Started: started, Ended: started.Add(time.Second)})
+	if err != nil {
+		panic(err)
+	}
+	return b
+}
+
 func pulseKeyOf(t *testing.T, bucket, key string) string {
 	t.Helper()
 	k, err := delivery.PulseKey(bucket, key)
@@ -207,7 +224,7 @@ func TestRebuildTakesThePresenceOfWhoIsGone(t *testing.T) { // 1135/AC2
 func TestRebuildPulsesFromCurrentKeys(t *testing.T) { // 1135/AC2
 	w, o, c := setup()
 	c.t = t0.Add(30 * time.Second)
-	w.Asked("go", "askid1", t0.Add(10*time.Second))
+	w.Step("go", "n1.askid1", runningStep("askid1", t0.Add(10*time.Second)), Put)
 	w.SeedPresence("ace", false, t0)
 	w.Reconcile()
 	if err := w.Rebuild(func([]byte) ([]byte, error) { return connzAnswer(), nil }); err != nil {
@@ -226,7 +243,7 @@ func TestReconcileDropsPulsesNothingMirrorsAnyMore(t *testing.T) { // 1135/AC2
 	live := pulseKeyOf(t, checksKey, "askid1")
 	w.SeedPulse(stale)
 	w.SeedPulse(live)
-	w.Asked("go", "askid1", t0)
+	w.Step("go", "n1.askid1", runningStep("askid1", t0), Put)
 	w.Reconcile()
 	if !slices.Equal(o.pulseDeletes, []string{stale}) {
 		t.Fatalf("deleted = %v, want only %s", o.pulseDeletes, stale)
@@ -239,7 +256,7 @@ func TestReconcileDropsPulsesNothingMirrorsAnyMore(t *testing.T) { // 1135/AC2
 func TestPulseTimeToLiveIsTheBoundMinusTheAge(t *testing.T) { // 1135/AC3
 	w, o, c := setup()
 	c.t = t0.Add(30 * time.Second)
-	w.Asked("go", "askid1", t0) // the key found at start, last renewed 30s ago
+	w.Step("go", "n1.askid1", runningStep("askid1", t0), Put) // the key found at start, last renewed 30s ago
 	k := pulseKeyOf(t, checksKey, "askid1")
 	if o.ttls[k] != 90*time.Second {
 		t.Fatalf("ttl = %v, want 90s (last renewed 30s ago)", o.ttls[k])
@@ -257,7 +274,7 @@ func TestPulseTimeToLiveIsTheBoundMinusTheAge(t *testing.T) { // 1135/AC3
 func TestAMoveAtTheBoundIsStuckAtOnce(t *testing.T) { // 1135/AC3
 	w, o, c := setup()
 	c.t = t0.Add(120 * time.Second)
-	w.Asked("go", "askid1", t0)
+	w.Step("go", "n1.askid1", runningStep("askid1", t0), Put)
 	if len(o.pulses) != 0 {
 		t.Fatalf("a pulse was written for a heartbeat 120s old: %v", o.pulses)
 	}
@@ -265,7 +282,7 @@ func TestAMoveAtTheBoundIsStuckAtOnce(t *testing.T) { // 1135/AC3
 	if !ok {
 		t.Fatalf("no stuck key; have %v", o.stuck)
 	}
-	if s.Moved || !s.Since.Equal(t0) || s.Waited != "the ask askid1 to be taken by a node" {
+	if s.Moved || !s.Since.Equal(t0) || s.Waited != "the ask askid1 to end on n1" {
 		t.Fatalf("stuck = %+v", s)
 	}
 }
@@ -273,13 +290,13 @@ func TestAMoveAtTheBoundIsStuckAtOnce(t *testing.T) { // 1135/AC3
 func TestWaitedInWords(t *testing.T) { // 1135/AC3
 	w, o, c := setup()
 	c.t = t0.Add(200 * time.Second)
-	w.Asked("go", "askid1", t0)
+	w.Step("go", "n1.askid1", runningStep("askid1", t0), Put)
 	w.SeedPresence("ace", false, t0)
 	w.SeedPresence("part.mesh-delivery-view.ace", false, t0)
 	w.Reconcile()
 	_ = w.Rebuild(func([]byte) ([]byte, error) { return connzAnswer(), nil })
 	want := map[string]string{
-		stuckKeyOf(t, checksKey, "askid1"):                       "the ask askid1 to be taken by a node",
+		stuckKeyOf(t, checksKey, "askid1"):                       "the ask askid1 to end on n1",
 		stuckKeyOf(t, presenceNS, "ace"):                         "ace to come back online",
 		stuckKeyOf(t, presenceNS, "part.mesh-delivery-view.ace"): "mesh-delivery-view on ace to come back online",
 	}
@@ -293,13 +310,13 @@ func TestWaitedInWords(t *testing.T) { // 1135/AC3
 func TestEveryHeartbeatRenewsThePulse(t *testing.T) { // 1135/AC3
 	w, o, c := setup()
 	k := pulseKeyOf(t, checksKey, "askid1")
-	w.Asked("go", "askid1", t0)
+	w.Step("go", "n1.askid1", runningStep("askid1", t0), Put)
 	if o.ttls[k] != 120*time.Second {
 		t.Fatalf("ttl = %v, want the full 120s", o.ttls[k])
 	}
 	o.ttls[k] = 0
 	c.t = t0.Add(100 * time.Second)
-	w.Asked("go", "askid1", c.t) // the builder re-puts the key: same check, started long ago
+	w.Step("go", "n1.askid1", runningStep("askid1", c.t), Put) // the builder re-puts the key: same check, started long ago
 	if o.ttls[k] != 120*time.Second {
 		t.Fatalf("a heartbeat left the pulse as it was: ttl = %v, want the full 120s again", o.ttls[k])
 	}
@@ -308,7 +325,7 @@ func TestEveryHeartbeatRenewsThePulse(t *testing.T) { // 1135/AC3
 	}
 	// The same revision heard twice changes nothing.
 	o.ttls[k] = 0
-	w.Asked("go", "askid1", c.t)
+	w.Step("go", "n1.askid1", runningStep("askid1", c.t), Put)
 	if o.ttls[k] != 0 {
 		t.Fatal("one revision rewrote the pulse twice")
 	}
@@ -317,10 +334,10 @@ func TestEveryHeartbeatRenewsThePulse(t *testing.T) { // 1135/AC3
 func TestMirroredMovesDeleteThePulse(t *testing.T) { // 1135/AC4
 	w, o, c := setup()
 	c.t = t0.Add(10 * time.Second)
-	w.Asked("go", "askid1", t0)
+	w.Step("go", "n1.askid1", runningStep("askid1", t0), Put)
 	w.Connect("node.ace")
 	w.Disconnect("node.ace")
-	w.Taken("go", "askid1") // the check ended
+	w.Step("go", "n1.askid1", nil, Delete) // the check ended
 	w.Connect("node.ace")
 	if len(o.pulses) != 0 {
 		t.Fatalf("pulses left: %v", o.pulses)
@@ -335,9 +352,9 @@ func TestMirroredMovesDeleteThePulse(t *testing.T) { // 1135/AC4
 
 func TestACheckWhoseBuilderStopsIsStuckFromItsLastHeartbeat(t *testing.T) { // 1135/AC4
 	w, o, c := setup()
-	w.Asked("go", "askid1", t0)
+	w.Step("go", "n1.askid1", runningStep("askid1", t0), Put)
 	c.t = t0.Add(50 * time.Second)
-	w.Asked("go", "askid1", c.t) // the last heartbeat
+	w.Step("go", "n1.askid1", runningStep("askid1", c.t), Put) // the last heartbeat
 	c.t = t0.Add(170 * time.Second)
 	w.Pulse(pulseKeyOf(t, checksKey, "askid1"), Expired, c.t)
 	s := o.stuck[stuckKeyOf(t, checksKey, "askid1")]
@@ -348,7 +365,7 @@ func TestACheckWhoseBuilderStopsIsStuckFromItsLastHeartbeat(t *testing.T) { // 1
 
 func TestAStuckKeyIsWrittenMovedNeverDeleted(t *testing.T) { // 1135/AC4
 	w, o, c := setup()
-	w.Asked("go", "askid1", t0)
+	w.Step("go", "n1.askid1", runningStep("askid1", t0), Put)
 	pk := pulseKeyOf(t, checksKey, "askid1")
 	c.t = t0.Add(120 * time.Second)
 	w.Pulse(pk, Expired, c.t)
@@ -357,24 +374,24 @@ func TestAStuckKeyIsWrittenMovedNeverDeleted(t *testing.T) { // 1135/AC4
 		t.Fatal("stuck written as moved at expiry")
 	}
 	c.t = t0.Add(300 * time.Second)
-	w.Taken("go", "askid1")
+	w.Step("go", "n1.askid1", nil, Delete)
 	s := o.stuck[sk]
-	if !s.Moved || !s.Since.Equal(t0) || s.Waited != "the ask askid1 to be taken by a node" {
+	if !s.Moved || !s.Since.Equal(t0) || s.Waited != "the ask askid1 to end on n1" {
 		t.Fatalf("stuck after the move = %+v, want the same key moved", s)
 	}
 	n := o.stuckWrites
-	w.Taken("go", "askid1")
+	w.Step("go", "n1.askid1", nil, Delete)
 	if o.stuckWrites != n {
 		t.Fatal("a stuck key already moved was written again")
 	}
 }
 
-func TestAnAskTakenAfterStuckMovesIt(t *testing.T) { // 1135/AC4
+func TestAnAskEndedAfterStuckMovesIt(t *testing.T) { // 1135/AC4
 	w, o, c := setup()
-	w.Asked("go", "askid1", t0)
+	w.Step("go", "n1.askid1", runningStep("askid1", t0), Put)
 	c.t = t0.Add(120 * time.Second)
 	w.Pulse(pulseKeyOf(t, checksKey, "askid1"), Expired, c.t)
-	w.Taken("go", "askid1") // a node took it at last
+	w.Step("go", "n1.askid1", nil, Delete) // the ask ended at last
 	if !o.stuck[stuckKeyOf(t, checksKey, "askid1")].Moved {
 		t.Fatal("an ask taken after it was stuck is still stuck")
 	}
@@ -398,23 +415,23 @@ func TestPresenceStuckThenBackOnline(t *testing.T) { // 1135/AC4
 
 func TestExpiryMarkerWritesStuck(t *testing.T) { // 1135/AC5
 	w, o, c := setup()
-	w.Asked("go", "askid1", t0)
+	w.Step("go", "n1.askid1", runningStep("askid1", t0), Put)
 	c.t = t0.Add(120 * time.Second)
 	w.Pulse(pulseKeyOf(t, checksKey, "askid1"), Expired, c.t)
 	s, ok := o.stuck[stuckKeyOf(t, checksKey, "askid1")]
 	if !ok {
 		t.Fatalf("no stuck key; have %v", o.stuck)
 	}
-	if s.Bucket != checksKey || s.Key != "askid1" || s.Waited != "the ask askid1 to be taken by a node" || !s.Since.Equal(t0) || s.Moved {
+	if s.Bucket != checksKey || s.Key != "askid1" || s.Waited != "the ask askid1 to end on n1" || !s.Since.Equal(t0) || s.Moved {
 		t.Fatalf("stuck = %+v", s)
 	}
 }
 
 func TestOwnDeleteOfAPulseWritesNothing(t *testing.T) { // 1135/AC5
 	w, o, c := setup()
-	w.Asked("go", "askid1", t0)
+	w.Step("go", "n1.askid1", runningStep("askid1", t0), Put)
 	pk := pulseKeyOf(t, checksKey, "askid1")
-	w.Taken("go", "askid1")
+	w.Step("go", "n1.askid1", nil, Delete)
 	w.Pulse(pk, Delete, c.t) // the delete the watchman made, heard on its own watch
 	w.Pulse(pk, Expired, c.t)
 	w.Pulse("mesh-delivery-watchman_pulse.unknown.key", Expired, c.t)
@@ -427,13 +444,56 @@ func TestTheBoundCanBeShortenedForATest(t *testing.T) { // 1135/AC8
 	w, o, c := setup()
 	w.SetBound(2 * time.Second)
 	c.t = t0.Add(500 * time.Millisecond)
-	w.Asked("go", "askid1", t0)
+	w.Step("go", "n1.askid1", runningStep("askid1", t0), Put)
 	if got := o.ttls[pulseKeyOf(t, checksKey, "askid1")]; got != 1500*time.Millisecond {
 		t.Fatalf("ttl = %v, want 1.5s", got)
 	}
 	c.t = t0.Add(3 * time.Second)
-	w.Asked("go", "askid2", t0)
+	w.Step("go", "n1.askid2", runningStep("askid2", t0), Put)
 	if _, ok := o.stuck[stuckKeyOf(t, checksKey, "askid2")]; !ok {
 		t.Fatalf("a heartbeat older than the bound is stuck at once; have %v", o.stuck)
 	}
 }
+
+func TestAnAskTakenByANodeHasAPulseFromWhenItStartedAndEndsWithTheStep(t *testing.T) { // 1908, 1909
+	w, o, c := setup()
+	c.t = t0.Add(10 * time.Second)
+	w.Step("go", "n1.askid1", runningStep("askid1", t0), Put)
+	if len(o.pulses) != 1 {
+		t.Fatalf("pulses = %v, want the taken ask's", o.pulses)
+	}
+	for _, state := range []string{delivery.StepPassed, delivery.StepFailed, delivery.StepRefused} {
+		w2, o2, c2 := setup()
+		c2.t = t0.Add(10 * time.Second)
+		w2.Step("go", "n1.askid1", runningStep("askid1", t0), Put)
+		w2.Step("go", "n1.askid1", endedStep("askid1", state, t0), Put)
+		if len(o2.pulses) != 0 && o2.pulses[pulseKeyOf(t, checksKey, "askid1")].Key != "" {
+			t.Errorf("%s: the pulse of an ended ask stands: %v", state, o2.pulses)
+		}
+		_ = c2
+	}
+	_ = c
+}
+
+func TestARunningAskFoundAtStartGetsItsPulse(t *testing.T) { // 1909
+	w, o, c := setup()
+	c.t = t0.Add(30 * time.Second)
+	w.Step("go", "n1.askid1", runningStep("askid1", t0), Put) // what the steps held when the watchman started
+	p, ok := o.pulses[pulseKeyOf(t, checksKey, "askid1")]
+	if !ok || p.Key != "askid1" {
+		t.Fatalf("pulses = %v", o.pulses)
+	}
+	if got := o.ttls[pulseKeyOf(t, checksKey, "askid1")]; got <= 0 || got >= 120*time.Second {
+		t.Errorf("ttl = %v, want the bound less its 30s", got)
+	}
+}
+
+func TestAPausedKeyOrAnotherNodesNonStepIsLeft(t *testing.T) { // 1908
+	w, o, _ := setup()
+	w.Step("go", "paused.n1", []byte(`{}`), Put)
+	w.Step("go", "n1.not-an-id!", []byte(`{}`), Put)
+	w.Step("go", "n1.askid1", []byte(`{"v":99}`), Put)
+	if len(o.pulses) != 0 || len(o.stuck) != 0 {
+		t.Errorf("pulses %v stuck %v", o.pulses, o.stuck)
+	}
+}