Merge pull request 'Builder: given-up asks no longer hold a re-ask back for ever, and a tool reads a build seat's queue (issue 1945)' (#56) from fix/1945-reask-dead-ahead into main

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

@@ -5,19 +5,26 @@
 package main
 
 import (
+	"context"
 	"encoding/json"
 	"fmt"
 	"os"
 	"strings"
+	"sync"
 	"time"
 
 	stdio "git.novox.be/novox/mesh-sdk/go"
+	"git.novox.be/novox/mesh-sdk/go/delivery"
 
 	"mesh-delivery-builder/internal/build"
 )
 
-// Tools are the builder's two tools: stuck reads, askAgain writes the reasks value.
+// queueAsker writes a queues value and waits for the builder's answer to it.
-func Tools(r build.StuckReader, askAgain func(key string, value []byte) error, now func() time.Time) []stdio.Tool {
+type queueAsker func(key string, value []byte) (json.RawMessage, error)
+
+// Tools are the builder's tools: stuck reads, askAgain writes the reasks value, queue writes a queues value and waits for
+// the builder's read.
+func Tools(r build.StuckReader, askAgain func(key string, value []byte) error, queue queueAsker, now func() time.Time) []stdio.Tool {
 	str := func(args map[string]any, k string) string { s, _ := args[k].(string); return strings.TrimSpace(s) }
 	return []stdio.Tool{
 		{Name: build.StuckTool,
@@ -48,6 +55,78 @@ func Tools(r build.StuckReader, askAgain func(key string, value []byte) error, n
 				return map[string]any{"key": key, "asked": true,
 					"next": "the builder answers in " + build.ReAsksState + "." + key + "; " + build.StuckTool + " shows the answer and the new ask"}, nil
 			}},
+		{Name: build.SeatQueueTool,
+			Description: "What waits on a build seat's stream (SEAT_NODE_BUILD_<LANGUAGE>), as the builder reads it: the stream's message count and " +
+				"sequences, and per kind (delivery, check, recheck) its messages oldest first, at most limit each, with sequence, ask id, when asked " +
+				"and put, age, the state of the step a build node wrote for it, and whether the builder counts it as still waiting or as given up " +
+				"on (its worker delivered it its most times) or taken; and the ask asked last that a node took. The workers' delivery counts are " +
+				"not readable by the builder (nats.nats_streams shows them). Read by the builder, which alone may read the stream; read-only.",
+			Input: map[string]any{"type": "object", "properties": map[string]any{
+				"language": map[string]any{"type": "string", "description": "go, node or image (default go)"},
+				"seat":     map[string]any{"type": "string", "description": "or the seat: node-build-go, node-build-node, node-build-image"},
+				"limit":    map[string]any{"type": "integer", "description": fmt.Sprintf("messages read of each kind (default %d, at most %d)", build.DefaultQueueLimit, build.MaxQueueLimit)}}},
+			Run: func(args map[string]any) (any, error) {
+				language := build.LanguageOf(str(args, "language"))
+				if language == "" {
+					language = build.LanguageOf(str(args, "seat"))
+				}
+				if language == "" {
+					language = delivery.AskGo
+				}
+				limit, _ := args["limit"].(float64)
+				key, value, err := build.NewQueueAsk(language, int(limit), stdio.CallerOf(args), now())
+				if err != nil {
+					return nil, err
+				}
+				return queue(key, value)
+			}},
+	}
+}
+
+// askQueue writes the queues value at key and waits, on a watch of the key, at most bound for the builder's answer.
+func askQueue(s stdio.KeptState, bound time.Duration) queueAsker {
+	return func(key string, value []byte) (json.RawMessage, error) {
+		answered := make(chan json.RawMessage, 1)
+		var rev uint64
+		var mu sync.Mutex
+		if err := s.Watch(key, func(c stdio.StateChange) error {
+			mu.Lock()
+			mine := rev
+			mu.Unlock()
+			if c.Op != "put" || mine == 0 || c.Revision <= mine {
+				return nil
+			}
+			if a, err := build.DecodeQueueAsk(c.Value); err == nil && !a.Answered.IsZero() {
+				select {
+				case answered <- c.Value:
+				default:
+				}
+			}
+			return nil
+		}); err != nil {
+			return nil, fmt.Errorf("watching %s.%s: %w", build.QueuesState, key, err)
+		}
+		r, err := s.Put(key, json.RawMessage(value))
+		if err != nil {
+			return nil, fmt.Errorf("the read could not be written to %s: %w", build.QueuesBucket, err)
+		}
+		mu.Lock()
+		rev = r
+		mu.Unlock()
+		// The watch may have heard the answer before the revision was known: read it once too.
+		if e, err := s.Get(key); err == nil && e != nil && e.Revision > r {
+			if a, err := build.DecodeQueueAsk(e.Value); err == nil && !a.Answered.IsZero() {
+				return e.Value, nil
+			}
+		}
+		wait, cancel := context.WithTimeout(context.Background(), bound) // a context's bound, no timer (ADR 0437 §8)
+		defer cancel()
+		select {
+		case v := <-answered:
+			return v, nil
+		case <-wait.Done():
+			return nil, fmt.Errorf("the builder did not answer %s.%s within %s: it is not running, or cannot read %s", build.QueuesState, key, bound, build.QueuesBucket)
+		}
 	}
 }
 
@@ -92,7 +171,7 @@ func main() {
 		_, err := reasks.Put(key, json.RawMessage(value))
 		return err
 	}
-	if err := stdio.Serve("", Tools(r, put, time.Now)); err != nil {
+	if err := stdio.Serve("", Tools(r, put, askQueue(stdio.State(build.QueuesState), 25*time.Second), time.Now)); err != nil {
 		fmt.Fprintf(os.Stderr, "[mesh-delivery-builder-tools] %v\n", err)
 		os.Exit(1)
 	}

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

@@ -4,6 +4,9 @@ import (
 	"context"
 	"errors"
 	"fmt"
+	"sort"
+	"strings"
+	"time"
 
 	"github.com/nats-io/nats.go/jetstream"
 
@@ -36,22 +39,29 @@ func (h heldOnTheBus) Gone(ctx context.Context, a delivery.StepAsk) (string, err
 	return seen.Why(), nil
 }
 
-// seen reads what the bus holds of the ask; with later, a step of its build module started after the ask was put too,
+// maxOnSubject bounds the messages of one subject a read walks (each a read by sequence, next by subject), and maxSteps
-// when the ask waits first in line.
+// the steps it reads whole to find the last ask taken: past them the read says it cannot tell, and the ask is waited for.
+const (
+	maxOnSubject = 5000
+	maxSteps     = 200
+)
+
+// seen reads what the bus holds of the ask; with later, the ask asked last that a build node took too (AskSeen.Last), when
+// the ask is on its stream.
 func (h heldOnTheBus) seen(ctx context.Context, a delivery.StepAsk, later bool) (build.AskSeen, error) {
-	var seen build.AskSeen
+	seen := build.AskSeen{Kind: a.Kind, At: build.AskedAt(a.ID)}
 	module, err := delivery.BuildModuleOf(a.Language)
 	if err != nil {
 		return seen, err
 	}
-	pattern, err := delivery.AskStepWatch(a.ID)
-	if err != nil {
-		return seen, err
-	}
 	steps, err := h.js.KeyValue(ctx, delivery.StepsBucket(module))
 	if err != nil {
 		return seen, fmt.Errorf("the bucket %s: %w", delivery.StepsBucket(module), err)
 	}
+	pattern, err := delivery.AskStepWatch(a.ID)
+	if err != nil {
+		return seen, err
+	}
 	keys, err := steps.ListKeysFiltered(ctx, pattern)
 	if err != nil {
 		return seen, err
@@ -61,67 +71,236 @@ func (h heldOnTheBus) seen(ctx context.Context, a delivery.StepAsk, later bool)
 		seen.Step = true // a node took it
 		return seen, nil
 	}
-	subject, err := delivery.AskSubject(a.Language, a.Kind)
+	stream, name, subject, err := h.stream(ctx, a.Language, a.Kind)
-	if err != nil {
-		return seen, err
-	}
-	// The seat's stream by its name (delivery.AskStream), as the broker grants the builder its information and its
-	// messages by sequence and nothing wider (issue 1910).
-	name, err := delivery.AskStream(a.Language)
-	if err != nil {
-		return seen, err
-	}
 	seen.Stream = name
-	stream, err := h.js.Stream(ctx, name)
 	if errors.Is(err, jetstream.ErrStreamNotFound) {
 		return seen, nil
 	}
 	if err != nil {
 		return seen, err
 	}
-	// A seat's stream is a work queue, which takes no second consumer over its subjects: its messages are read by sequence
+	var ahead []build.Queued
-	// instead, each a direct read that consumes nothing.
+	err = onSubject(ctx, stream, subject, func(q build.Queued) bool {
-	info, err := stream.Info(ctx)
+		if q.Ask == a.ID {
+			seen.OnStream, seen.PutAt = true, q.PutAt
+			return false
+		}
+		ahead = append(ahead, q)
+		return true
+	})
+	if err != nil || !seen.OnStream || !later {
+		return seen, err
+	}
+	taken, err := stepKeys(ctx, steps)
 	if err != nil {
 		return seen, err
 	}
-	for seq := info.State.FirstSeq; seq != 0 && seq <= info.State.LastSeq; seq++ {
+	for i := range ahead {
-		m, err := stream.GetMsg(ctx, seq)
+		ahead[i].Step = stepState(ctx, steps, taken[ahead[i].Ask])
+	}
+	seen.Ahead = ahead
+	since := seen.At
+	if since.IsZero() {
+		since = seen.PutAt
+	}
+	for _, q := range ahead {
+		if q.At.Before(since) {
+			since = q.At
+		}
+	}
+	seen.Last = lastTaken(ctx, steps, taken, a.Kind, since)
+	return seen, nil
+}
+
+// stream is the build seat's stream of language, its name and the subject of kind on it.
+func (h heldOnTheBus) stream(ctx context.Context, language, kind string) (jetstream.Stream, string, string, error) {
+	subject, err := delivery.AskSubject(language, kind)
+	if err != nil {
+		return nil, "", "", err
+	}
+	// The seat's stream by its name (delivery.AskStream), as the broker grants the builder its information and its
+	// messages by sequence and nothing wider (issue 1910).
+	name, err := delivery.AskStream(language)
+	if err != nil {
+		return nil, "", "", err
+	}
+	stream, err := h.js.Stream(ctx, name)
+	return stream, name, subject, err
+}
+
+// onSubject hands each message of subject on stream to each, in order, until each answers false or the subject's
+// messages end. A seat's stream is a work queue, which takes no second consumer over its subjects: its messages are read
+// by sequence instead, each the next of the subject (one read per message, none for the sequences taken between), and
+// none consumed.
+func onSubject(ctx context.Context, stream jetstream.Stream, subject string, each func(build.Queued) bool) error {
+	for seq, n := uint64(1), 0; ; n++ {
+		if n >= maxOnSubject {
+			return fmt.Errorf("%s holds more than %d messages before the one asked after: not read further", subject, maxOnSubject)
+		}
+		m, err := stream.GetMsg(ctx, seq, jetstream.WithGetMsgSubject(subject))
 		if errors.Is(err, jetstream.ErrMsgNotFound) {
-			continue // taken and acknowledged, or removed
+			return nil
 		}
 		if err != nil {
-			return seen, err
+			return err
 		}
-		if m.Subject != subject {
+		q := build.Queued{Seq: m.Sequence, Ask: m.Header.Get(jetstream.MsgIDHeader), PutAt: m.Time}
-			continue
+		if _, kind, err := delivery.ParseAskSubject(m.Subject); err == nil {
+			q.Kind = kind
 		}
-		if m.Header.Get(jetstream.MsgIDHeader) == a.ID {
+		if q.At = build.AskedAt(q.Ask); q.At.IsZero() {
-			seen.OnStream, seen.PutAt = true, m.Time
+			q.At = q.PutAt
-			break
 		}
-		seen.Ahead++
+		if !each(q) {
+			return nil
+		}
+		seq = m.Sequence + 1
 	}
-	if !later || !seen.OnStream || seen.Ahead > 0 {
+}
-		return seen, nil
+
-	}
+// stepKeys are the build module's step keys by ask id: `<node>.<ask id>` (a node's own key, such as its pause, has no
-	// First in line and not taken: did a build node take any ask put after it? Its stream then no longer delivers it.
+// ask id and is left out).
+func stepKeys(ctx context.Context, steps jetstream.KeyValue) (map[string][]string, error) {
 	all, err := steps.ListKeys(ctx)
 	if err != nil {
-		return seen, err
+		if errors.Is(err, jetstream.ErrNoKeysFound) {
+			return map[string][]string{}, nil
+		}
+		return nil, err
 	}
 	defer all.Stop()
+	out := map[string][]string{}
 	for k := range all.Keys() {
+		if _, id, ok := strings.Cut(k, "."); ok && !build.AskedAt(id).IsZero() {
+			out[id] = append(out[id], k)
+		}
+	}
+	return out, nil
+}
+
+// stepState is the state of the first of keys a node wrote, "" for none.
+func stepState(ctx context.Context, steps jetstream.KeyValue, keys []string) string {
+	for _, k := range keys {
 		e, err := steps.Get(ctx, k)
 		if err != nil {
 			continue
 		}
-		s, err := delivery.DecodeStep(e.Value())
+		if s, err := delivery.DecodeStep(e.Value()); err == nil {
-		if err != nil || !s.Started.After(seen.PutAt) {
+			return s.State
-			continue // not a step (a node's pause key), or one started before the ask was put
 		}
-		seen.Later = &s
-		break
 	}
-	return seen, nil
+	return ""
+}
+
+// lastTaken is the ask asked last after since that a build node took, of a kind whose taking proves kind passed
+// (build.Proves), read newest first and at most maxSteps of them; nil for none. A step's tool is "<kind> ask"
+// (mesh-sdk buildstep).
+func lastTaken(ctx context.Context, steps jetstream.KeyValue, taken map[string][]string, kind string, since time.Time) *build.Taken {
+	var ids []string
+	for id := range taken {
+		if build.AskedAt(id).After(since) {
+			ids = append(ids, id)
+		}
+	}
+	sort.Slice(ids, func(i, j int) bool { return build.AskedAt(ids[i]).After(build.AskedAt(ids[j])) })
+	read := 0
+	for _, id := range ids {
+		for _, k := range taken[id] {
+			if read >= maxSteps {
+				return nil
+			}
+			read++
+			e, err := steps.Get(ctx, k)
+			if err != nil {
+				continue
+			}
+			s, err := delivery.DecodeStep(e.Value())
+			if err != nil {
+				continue
+			}
+			stepKind, ok := strings.CutSuffix(s.Tool, " ask")
+			if !ok || !build.Proves(stepKind, kind) {
+				continue
+			}
+			return &build.Taken{Ask: id, Kind: stepKind, At: build.AskedAt(id), Node: s.Node, Started: s.Started}
+		}
+	}
+	return nil
+}
+
+// Drop is build.Dropper's Drop: it removes the ask's message from its build seat's stream, found by its Nats-Msg-Id on its
+// subject, when the stream still holds it. The builder's account must be granted $JS.API.STREAM.MSG.DELETE.<stream>.
+func (h heldOnTheBus) Drop(ctx context.Context, a delivery.StepAsk) error {
+	stream, name, subject, err := h.stream(ctx, a.Language, a.Kind)
+	if errors.Is(err, jetstream.ErrStreamNotFound) {
+		return nil
+	}
+	if err != nil {
+		return err
+	}
+	var seq uint64
+	if err := onSubject(ctx, stream, subject, func(q build.Queued) bool {
+		if q.Ask == a.ID {
+			seq = q.Seq
+			return false
+		}
+		return true
+	}); err != nil || seq == 0 {
+		return err
+	}
+	if err := stream.DeleteMsg(ctx, seq); err != nil && !errors.Is(err, jetstream.ErrMsgNotFound) {
+		return fmt.Errorf("deleting message %d of %s (is $JS.API.STREAM.MSG.DELETE.%s granted to the builder?): %w", seq, name, name, err)
+	}
+	return nil
+}
+
+// Queue is what the tool builder_seat_queue answers (build.SeatQueue): the messages of language's build seat's stream, of
+// every kind, oldest first and at most limit of each kind, each with its step's state and whether it still waits.
+func (h heldOnTheBus) Queue(ctx context.Context, language string, limit int) (build.SeatQueue, error) {
+	q := build.SeatQueue{Language: language, Note: build.QueueNote}
+	module, err := delivery.BuildModuleOf(language)
+	if err != nil {
+		return q, err
+	}
+	stream, name, _, err := h.stream(ctx, language, delivery.AskDelivery)
+	q.Stream = name
+	if errors.Is(err, jetstream.ErrStreamNotFound) {
+		q.Note = name + " does not exist"
+		return q, nil
+	}
+	if err != nil {
+		return q, err
+	}
+	info, err := stream.Info(ctx)
+	if err != nil {
+		return q, err
+	}
+	q.Messages, q.FirstSeq, q.LastSeq = info.State.Msgs, info.State.FirstSeq, info.State.LastSeq
+	steps, err := h.js.KeyValue(ctx, delivery.StepsBucket(module))
+	if err != nil {
+		return q, fmt.Errorf("the bucket %s: %w", delivery.StepsBucket(module), err)
+	}
+	taken, err := stepKeys(ctx, steps)
+	if err != nil {
+		return q, err
+	}
+	for _, kind := range delivery.AskKinds {
+		subject, _ := delivery.AskSubject(language, kind)
+		var ahead []build.Queued
+		if err := onSubject(ctx, stream, subject, func(m build.Queued) bool {
+			ahead = append(ahead, m)
+			return len(ahead) < limit
+		}); err != nil {
+			return q, err
+		}
+		if len(ahead) == 0 {
+			continue
+		}
+		for i := range ahead {
+			ahead[i].Step = stepState(ctx, steps, taken[ahead[i].Ask])
+		}
+		seen := build.AskSeen{Kind: kind, Ahead: ahead, Last: lastTaken(ctx, steps, taken, kind, ahead[0].At)}
+		q.Kinds = append(q.Kinds, build.NewKindQueue(kind, seen, time.Now()))
+	}
+	return q, nil
 }

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

@@ -24,6 +24,8 @@ import (
 
 	"git.novox.be/novox/mesh-sdk/go/buildstep"
 	"git.novox.be/novox/mesh-sdk/go/delivery"
+
+	"mesh-delivery-builder/internal/build"
 )
 
 // busConf is the shape of the composed accounts file (mesh-controller internal/broker): one account with JetStream, the
@@ -822,6 +824,34 @@ func TestARestartedBuilderPutsNoSecondAsk(t *testing.T) {
 			t.Fatalf("asks put = %+v within 20 s, want the two the key names", asked())
 		}
 	}
+	// 1945 note 62, on the real bus: the stream read as builder_seat_queue reads it shows the Go ask waiting; the ask can
+	// still be answered (nothing was taken since); and a message dropped as an ask asked again is gone from the stream.
+	held := heldOnTheBus{b.js}
+	goAsk := delivery.StepAsk{ID: v.Asks[delivery.AskGo], Language: delivery.AskGo, Kind: delivery.AskDelivery}
+	stale := delivery.StepAsk{ID: "mvo53jcm70b5f7", Language: delivery.AskGo, Kind: delivery.AskDelivery}
+	subject, _ := delivery.AskSubject(delivery.AskGo, delivery.AskDelivery)
+	if _, err := b.js.Publish(b.ctx, subject, []byte("{}"), jetstream.WithMsgID(stale.ID)); err != nil {
+		t.Fatal(err)
+	}
+	queued := func() []build.QueuedState {
+		q, err := held.Queue(b.ctx, delivery.AskGo, 10)
+		if err != nil || len(q.Kinds) != 1 || q.Kinds[0].Kind != delivery.AskDelivery {
+			t.Fatalf("the Go seat's queue = %+v, %v; want its delivery asks", q, err)
+		}
+		return q.Kinds[0].Messages
+	}
+	if m := queued(); len(m) != 2 || m[0].Ask != goAsk.ID || !m[0].Waits || m[1].Ask != stale.ID {
+		t.Fatalf("the Go delivery queue = %+v, want the key's ask waiting, then the stale one", m)
+	}
+	if why, err := held.Gone(b.ctx, goAsk); why != "" || err != nil {
+		t.Errorf("Gone(the waiting ask) = %q, %v; want it answerable", why, err)
+	}
+	if err := held.Drop(b.ctx, stale); err != nil {
+		t.Fatal(err)
+	}
+	if m := queued(); len(m) != 1 || m[0].Ask != goAsk.ID {
+		t.Fatalf("after the drop the Go delivery queue = %+v, want the key's ask alone", m)
+	}
 	b.stop()
 	b.start(t, "n1", nil) // the same builder, started again, while both asks wait
 	b.fake.open()

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

@@ -176,7 +176,7 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
 	// revision (the gate's source state). The source's revisions are watched with their history (a direct get of one revision
 	// is not what a read grants): a revision heard after the candidate that names it builds that candidate then. Where it
 	// cannot be read the trunk's keys are built and the candidates wait (issue 1945).
-	errs := make(chan error, 4)
+	errs := make(chan error, 6)
 	trunks := &trunkSource{ctx: ctx, js: js, heads: &sourceHeads{}, moved: b.TrunkMoved, errs: errs, logf: logger.Printf}
 	trunks.start()
 	b.Trunk = trunks.head
@@ -288,6 +288,16 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
 			go func() { errs <- follow(ctx, rw, eachReAsk) }()
 		}
 	}
+	// A person reads a build seat's stream with builder_seat_queue, which writes the queues state (issue 1945 note 62): the
+	// builder alone may read the stream, so it answers there.
+	if kv := process.OptionalBucket(ctx, js, build.QueuesBucket, "builder_seat_queue is not answered until the builder starts again", logger.Printf); kv != nil {
+		eachRead := func(e jetstream.KeyValueEntry) { queueRead(ctx, heldOnTheBus{js}, kv, e, logger.Printf) }
+		if qw, err := settle(ctx, kv, eachRead); err != nil {
+			process.NotWatched(logger.Printf, build.QueuesBucket, err, "builder_seat_queue is not answered until the builder starts again")
+		} else {
+			go func() { errs <- follow(ctx, qw, eachRead) }()
+		}
+	}
 	logger.Printf("started on %s", cred.Node)
 	go func() { errs <- follow(ctx, w, each) }()
 	select {

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

@@ -0,0 +1,49 @@
+package main
+
+import (
+	"context"
+	"encoding/json"
+	"errors"
+	"time"
+
+	"github.com/nats-io/nats.go/jetstream"
+
+	"mesh-delivery-builder/internal/build"
+)
+
+// queueReader is the part of the builder a queues value reads through: heldOnTheBus.
+type queueReader interface {
+	Queue(ctx context.Context, language string, limit int) (build.SeatQueue, error)
+}
+
+// queueRead answers one value of the queues state (the tool builder_seat_queue, issue 1945 note 62): a value not yet
+// answered is read now, from the stream the builder's account may read, and written back by compare-and-set at its
+// revision with what was read or why it could not be. A value answered already (this builder's own write, or another
+// instance's) or deleted is left.
+func queueRead(ctx context.Context, r queueReader, kv jetstream.KeyValue, e jetstream.KeyValueEntry, logf func(string, ...any)) {
+	if e.Operation() != jetstream.KeyValuePut {
+		return
+	}
+	a, err := build.DecodeQueueAsk(e.Value())
+	if err != nil {
+		logf("%s.%s holds what is not a queue read, left: %v", build.QueuesState, e.Key(), err)
+		return
+	}
+	if !a.Answered.IsZero() {
+		return
+	}
+	read, cancel := context.WithTimeout(ctx, 20*time.Second)
+	q, err := r.Queue(read, e.Key(), max(1, min(a.Limit, build.MaxQueueLimit)))
+	cancel()
+	if err != nil {
+		a.Error = err.Error()
+	}
+	a.Queue, a.Answered = &q, time.Now().UTC()
+	data, err := json.Marshal(a)
+	if err != nil {
+		return
+	}
+	if _, err := kv.Update(ctx, e.Key(), data, e.Revision()); err != nil && !errors.Is(err, jetstream.ErrKeyExists) && !errors.Is(err, jetstream.ErrKeyRevisionMismatch) {
+		logf("%s.%s: the read could not be written: %v", build.QueuesState, e.Key(), err)
+	}
+}

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

@@ -5,6 +5,7 @@ import (
 	"errors"
 	"fmt"
 	"regexp"
+	"slices"
 	"strconv"
 	"strings"
 	"time"
@@ -44,33 +45,133 @@ type AskedAgain struct {
 
 // AskSeen is what the bus says of an ask whose step has not ended, as the builder may read it: its build seat's stream
 // (information and messages by sequence) and its build module's steps. The builder may not read the seat workers'
-// delivery counts (granted to the delivery view alone), so a message given up on is told by what the workers did after it.
+// delivery counts or ack floors (granted to the delivery view alone: $JS.API.CONSUMER.INFO.<stream>.<worker>), so a
+// message given up on is told by what the workers did after it: a build node took an ask asked later, of the same kind or
+// of one pulled after it (issue 1945 note 62: seven messages SEAT_NODE_BUILD_GO had given up on sat ahead of a Go ask for
+// five hours, each counted as waiting, so the ask was never asked again).
 type AskSeen struct {
 	Stream string
+	// Kind is the ask's kind, At when it was asked (its id's time).
+	Kind string
+	At   time.Time
 	// Step is whether a build node wrote a step for the ask (running: the step bound of buildrun.Asker judges it).
 	Step bool
-	// OnStream is whether the stream still holds the ask's message; PutAt is when it was put, Ahead how many messages of
+	// OnStream is whether the stream still holds the ask's message; PutAt is when it was put.
-	// its subject wait before it.
 	OnStream bool
 	PutAt    time.Time
-	Ahead    int
+	// Ahead are the messages of its subject before it on the stream.
-	// Later is a step of the same build module that started after the ask was put, when one did.
+	Ahead []Queued
-	Later *delivery.Step
+	// Last is the ask asked last that a build node took (wrote a step for) of the ask's kind or of a kind pulled after
+	// it, when one was taken: every message of the kind asked before it was then delivered already, or given up on.
+	Last *Taken
 }
 
-// Why is why the ask can no longer be answered, or "" when it can: it has a step, waits behind others of its kind, or no
+// Queued is one message of a build seat's stream: its sequence, its ask's id and kind, when the ask was asked (its id's
-// build node took any ask since it was put (the nodes are busy or paused: a backlog, never a fault, ADR 0525 §9).
+// time, the put time when the id says none) and put, and the state of the step a build node wrote for it ("" for none).
+type Queued struct {
+	Seq   uint64    `json:"seq"`
+	Ask   string    `json:"ask"`
+	Kind  string    `json:"kind"`
+	At    time.Time `json:"asked"`
+	PutAt time.Time `json:"put"`
+	Step  string    `json:"step,omitempty"`
+}
+
+// Taken is an ask a build node took: its id, kind and when it was asked, and the step's node and start.
+type Taken struct {
+	Ask     string    `json:"ask"`
+	Kind    string    `json:"kind"`
+	At      time.Time `json:"asked"`
+	Node    string    `json:"node"`
+	Started time.Time `json:"started"`
+}
+
+// Proves is whether a taken ask of kind taken shows that a message of kind waiting asked before it is no longer
+// delivered: a holder pulls the kinds in delivery.AskKinds' order, so taking one of the same kind or of a later one means
+// its worker had none of waiting's kind left to give.
+func Proves(taken, waiting string) bool {
+	t, w := slices.Index(delivery.AskKinds, taken), slices.Index(delivery.AskKinds, waiting)
+	return t >= 0 && w >= 0 && t >= w
+}
+
+// Waits is whether a message ahead still waits to be delivered: no build node wrote a step for it, and no ask of its
+// kind asked after it was taken. A message its worker gave up on (delivered its most times) stays on a work queue until
+// its age, and is not counted.
+func (s AskSeen) Waits(q Queued) bool {
+	return q.Step == "" && !s.passed(q.At)
+}
+
+func (s AskSeen) passed(at time.Time) bool {
+	return s.Last != nil && Proves(s.Last.Kind, s.Kind) && s.Last.At.After(at)
+}
+
+// Waiting is how many messages ahead still wait to be delivered (Waits).
+func (s AskSeen) Waiting() int {
+	n := 0
+	for _, q := range s.Ahead {
+		if s.Waits(q) {
+			n++
+		}
+	}
+	return n
+}
+
+// Why is why the ask can no longer be answered, or "" when it can: it has a step, waits behind others of its kind still
+// deliverable, or no build node took a later ask of its kind (the nodes are busy or paused: a backlog, never a fault,
+// ADR 0525 §9).
 func (s AskSeen) Why() string {
 	switch {
 	case s.Step:
 		return ""
 	case !s.OnStream:
 		return fmt.Sprintf("no build node wrote a step for it and %s no longer holds it: it was given up on or dropped", s.Stream)
-	case s.Ahead > 0 || s.Later == nil:
+	case s.Waiting() > 0 || !s.passed(s.at()):
 		return ""
 	default:
-		return fmt.Sprintf("it waits first in line on %s since %s while %s took a later ask at %s: its stream no longer delivers it (given up after its most deliveries)",
+		behind := ""
-			s.Stream, s.PutAt.UTC().Format(time.RFC3339), s.Later.Node, s.Later.Started.UTC().Format(time.RFC3339))
+		if n := len(s.Ahead); n > 0 {
+			behind = fmt.Sprintf(" behind %d given up on or taken", n)
+		}
+		return fmt.Sprintf("it waits on %s since %s%s while %s took the %s ask %s asked later at %s: its stream no longer delivers it (given up after its most deliveries)",
+			s.Stream, s.PutAt.UTC().Format(time.RFC3339), behind, s.Last.Node, s.Last.Kind, s.Last.Ask, s.Last.At.UTC().Format(time.RFC3339))
+	}
+}
+
+func (s AskSeen) at() time.Time {
+	if s.At.IsZero() {
+		return s.PutAt
+	}
+	return s.At
+}
+
+// AskedAt is when an ask id was made (buildstep.NewID: its milliseconds in base 36 before six hex digits), or zero.
+func AskedAt(id string) time.Time {
+	if len(id) <= 6 {
+		return time.Time{}
+	}
+	ms, err := strconv.ParseInt(id[:len(id)-6], 36, 64)
+	if err != nil || ms <= 0 {
+		return time.Time{}
+	}
+	return time.UnixMilli(ms)
+}
+
+// Dropper removes an ask's message from its build seat's stream (Builder.Gone may be one): the old message of an ask asked
+// again, so a message given up on does not sit ahead of every later ask of its kind until its age (issue 1945 note 62).
+type Dropper interface {
+	Drop(ctx context.Context, a delivery.StepAsk) error
+}
+
+// drop removes the old ask's message when Gone can; a refusal is said and changes nothing else: the new ask is put.
+func (b *Builder) drop(ctx context.Context, ask delivery.StepAsk, old string) {
+	d, ok := b.Gone.(Dropper)
+	if !ok {
+		return
+	}
+	prior := ask
+	prior.ID = old
+	if err := d.Drop(ctx, prior); err != nil {
+		b.logf("the message of the ask %s asked again as %s stays on its stream: %v", old, ask.ID, err)
 	}
 }
 
@@ -183,6 +284,7 @@ func (b *Builder) answer(ctx context.Context, l *lane, ask delivery.StepAsk, ask
 			if err := b.Asks.Put(ctx, ask); err != nil {
 				return step, ask, err
 			}
+			b.drop(ctx, ask, old)
 			b.logf("%s: the ask %s of %s is asked again as %s (asking %d of %d): %s", key, old, l.module, ask.ID, asking, b.askings(), why)
 			if b.Said != nil {
 				b.Said(AskedAgain{Key: key, Module: module, FP: fp, BuildModule: l.module, Language: ask.Language, Old: old, New: ask.ID,

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

@@ -0,0 +1,115 @@
+package build
+
+import (
+	"encoding/json"
+	"fmt"
+	"strings"
+	"time"
+
+	"git.novox.be/novox/mesh-sdk/go/delivery"
+)
+
+// **The tool builder_seat_queue reads a build seat's stream as the builder reads it** (novox/hq issue 1945 note 62; the
+// instruction "tools are how a module is spoken to"): the messages waiting on SEAT_NODE_BUILD_<LANGUAGE>, bounded, each
+// with its ask, its age, the state of the step a build node wrote for it, and whether the builder counts it as still
+// waiting (AskSeen.Waits) or as given up on. Only the builder's own account may read the stream (issue 1910): the tools
+// bundle writes the read into the builder's state queues, keyed by the language, which the builder watches and answers
+// there, as builder_ask_again does through reasks.
+const (
+	SeatQueueTool = "builder_seat_queue"
+	// QueuesState is the builder's state of queue reads: the bucket mesh-delivery-builder_queues, keyed by the language.
+	QueuesState  = "queues"
+	QueuesBucket = delivery.BuilderModule + "_" + QueuesState
+	// DefaultQueueLimit and MaxQueueLimit bound the messages read of each kind.
+	DefaultQueueLimit = 50
+	MaxQueueLimit     = 500
+)
+
+// QueueNote says what the read cannot show.
+const QueueNote = "the builder may not read the seat workers' delivery counts or ack floors ($JS.API.CONSUMER.INFO.<stream>.<worker> is " +
+	"granted to the delivery view alone): a message counts as given up on when no node wrote its step and a node took an ask of its kind " +
+	"asked after it; nats.nats_streams shows the workers' pending and waiting pulls"
+
+// QueueAsk is a value of the queues state: a read asked (limit, who, when) and, once the builder read the stream, its answer.
+type QueueAsk struct {
+	V        int        `json:"v"`
+	Limit    int        `json:"limit"`
+	By       string     `json:"by,omitempty"`
+	At       time.Time  `json:"at"`
+	Answered time.Time  `json:"answered,omitzero"`
+	Error    string     `json:"error,omitempty"`
+	Queue    *SeatQueue `json:"queue,omitempty"`
+}
+
+// SeatQueue is what the builder read of one build seat's stream.
+type SeatQueue struct {
+	Language string      `json:"language"`
+	Stream   string      `json:"stream"`
+	Messages uint64      `json:"messages"`
+	FirstSeq uint64      `json:"first_seq"`
+	LastSeq  uint64      `json:"last_seq"`
+	Kinds    []KindQueue `json:"kinds"`
+	Note     string      `json:"note"`
+}
+
+// KindQueue is one kind's messages on the stream, oldest first: how many the builder counts as waiting, the ask asked
+// last a build node took of the kind (or of one pulled after it), and each message.
+type KindQueue struct {
+	Kind     string        `json:"kind"`
+	Read     int           `json:"read"`
+	Waiting  int           `json:"waiting"`
+	GivenUp  int           `json:"given_up_or_taken"`
+	Last     *Taken        `json:"last_taken,omitempty"`
+	Messages []QueuedState `json:"messages"`
+}
+
+// QueuedState is one message as the tool shows it.
+type QueuedState struct {
+	Queued
+	AgeSeconds int64 `json:"age_s"`
+	Waits      bool  `json:"waits"`
+}
+
+// NewKindQueue is seen's messages ahead as one kind's queue at now.
+func NewKindQueue(kind string, seen AskSeen, now time.Time) KindQueue {
+	k := KindQueue{Kind: kind, Read: len(seen.Ahead), Last: seen.Last}
+	for _, q := range seen.Ahead {
+		w := seen.Waits(q)
+		if w {
+			k.Waiting++
+		} else {
+			k.GivenUp++
+		}
+		k.Messages = append(k.Messages, QueuedState{Queued: q, AgeSeconds: int64(now.Sub(q.PutAt) / time.Second), Waits: w})
+	}
+	return k
+}
+
+// NewQueueAsk is the value builder_seat_queue writes for a language: its key and the value.
+func NewQueueAsk(language string, limit int, by string, at time.Time) (string, []byte, error) {
+	if _, err := delivery.AskStream(language); err != nil {
+		return "", nil, fmt.Errorf("language: %w", err)
+	}
+	if limit <= 0 {
+		limit = DefaultQueueLimit
+	}
+	data, err := json.Marshal(QueueAsk{V: 1, Limit: min(limit, MaxQueueLimit), By: by, At: at.UTC()})
+	return language, data, err
+}
+
+// DecodeQueueAsk reads a queues value.
+func DecodeQueueAsk(data []byte) (QueueAsk, error) {
+	var a QueueAsk
+	if err := json.Unmarshal(data, &a); err != nil {
+		return a, err
+	}
+	if a.V != 1 {
+		return a, fmt.Errorf("a queues value of version %d, not 1", a.V)
+	}
+	return a, nil
+}
+
+// LanguageOf is the language a tool's seat or language argument names: go, node or image, or node-build-<language>.
+func LanguageOf(arg string) string {
+	return strings.TrimPrefix(strings.TrimSpace(arg), "node-build-")
+}

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

@@ -6,6 +6,7 @@ import (
 	"errors"
 	"fmt"
 	"git.novox.be/novox/mesh-sdk/go/delivery/buildrun"
+	"strconv"
 	"strings"
 	"sync"
 	"sync/atomic"
@@ -617,11 +618,20 @@ func TestAFailedPublishOfACandidatesImageIsWrittenToTheKey(t *testing.T) { // re
 	}
 }
 
-// goneAsks answers Gone from a function of the ask, and counts how often it was asked.
+// goneAsks answers Gone from a function of the ask, and counts how often it was asked; it keeps the ids it was asked to
+// drop from their stream (Dropper).
 type goneAsks struct {
-	mu  sync.Mutex
+	mu      sync.Mutex
-	n   int
+	n       int
-	why func(a delivery.StepAsk) string
+	why     func(a delivery.StepAsk) string
+	dropped []string
+}
+
+func (g *goneAsks) Drop(_ context.Context, a delivery.StepAsk) error {
+	g.mu.Lock()
+	defer g.mu.Unlock()
+	g.dropped = append(g.dropped, a.ID)
+	return nil
 }
 
 func (g *goneAsks) Gone(_ context.Context, a delivery.StepAsk) (string, error) {
@@ -703,14 +713,15 @@ func TestABuildAskWaitingInItsStreamIsNotBounded(t *testing.T) {
 					Step: c.step, Asks: map[string]string{delivery.AskGo: first}})
 				r.st.vals[r.key] = data
 			}
+			gone := &goneAsks{}
 			if !c.byHand {
-				r.b.Gone = &goneAsks{why: func(a delivery.StepAsk) string {
+				gone.why = func(a delivery.StepAsk) string {
 					if c.always || a.ID == first || first == "" {
 						return "it waits first in line while a node took a later ask"
 					}
 					return ""
-				}}
+				}
-				r.b.CheckAfter = 5 * time.Millisecond
+				r.b.Gone, r.b.CheckAfter = gone, 5*time.Millisecond
 			}
 			r.b.Module(context.Background(), "m", moduleValue(t, "modules/m"))
 			if c.byHand {
@@ -745,6 +756,11 @@ func TestABuildAskWaitingInItsStreamIsNotBounded(t *testing.T) {
 			}
 			mu.Lock()
 			defer mu.Unlock()
+			// Note 62: the old message of an ask asked again because its stream gave it up is dropped from the stream.
+			if gone.mu.Lock(); c.done && !c.byHand && (len(gone.dropped) != 1 || gone.dropped[0] != first) {
+				t.Errorf("dropped %v from the stream, want the old ask %s", gone.dropped, first)
+			}
+			gone.mu.Unlock()
 			if c.asking > 0 && (len(said) != 1 || said[0].Asking != c.asking || said[0].New == said[0].Old || said[0].Key != r.key || said[0].Why == "") {
 				t.Errorf("said %+v, want once at asking %d", said, c.asking)
 			}
@@ -752,22 +768,49 @@ func TestABuildAskWaitingInItsStreamIsNotBounded(t *testing.T) {
 	}
 	t.Run("what the bus says of an ask", func(t *testing.T) { // 1945
 		put := time.Date(2026, 10, 25, 17, 6, 0, 0, time.UTC)
-		later := &delivery.Step{Node: "g14", Started: put.Add(time.Minute)}
+		taken := func(kind string, at time.Time) *Taken {
+			return &Taken{Ask: "later", Kind: kind, At: at, Node: "g14", Started: at}
+		}
+		later := taken(delivery.AskDelivery, put.Add(time.Minute))
+		// ahead is n messages put before the ask, a minute apart, each with step (its state, "" for none).
+		ahead := func(n int, step string) []Queued {
+			var qs []Queued
+			for i := range n {
+				at := put.Add(-time.Duration(n-i) * time.Minute)
+				qs = append(qs, Queued{Seq: uint64(i + 1), Ask: fmt.Sprint("a", i), Kind: delivery.AskDelivery, At: at, PutAt: at, Step: step})
+			}
+			return qs
+		}
+		seen := func(qs []Queued, last *Taken) AskSeen {
+			return AskSeen{Stream: "S", Kind: delivery.AskDelivery, At: put, OnStream: true, PutAt: put, Ahead: qs, Last: last}
+		}
 		for _, c := range []struct {
-			seen AskSeen
+			name    string
-			want string // a part of the why, "" when it can be answered
+			seen    AskSeen
+			want    string // a part of the why, "" when it can be answered
+			waiting int
 		}{
-			{AskSeen{Stream: "S", Step: true}, ""},
+			{"a node took it", AskSeen{Stream: "S", Step: true}, "", 0},
-			{AskSeen{Stream: "S"}, "no longer holds it"},
+			{"no longer on its stream", AskSeen{Stream: "S"}, "no longer holds it", 0},
-			{AskSeen{Stream: "S", OnStream: true, PutAt: put, Later: later}, "no longer delivers it"},
+			{"first in line, a later ask taken", seen(nil, later), "no longer delivers it", 0},
-			{AskSeen{Stream: "S", OnStream: true, PutAt: put, Ahead: 2, Later: later}, ""},
+			{"first in line, nothing taken since", seen(nil, nil), "", 0},
-			{AskSeen{Stream: "S", OnStream: true, PutAt: put}, ""},
+			{"first in line, the ask taken last was asked before it", seen(nil, taken(delivery.AskDelivery, put.Add(-time.Second))), "", 0},
+			{"two wait ahead, nothing taken since", seen(ahead(2, ""), nil), "", 2},
+			{"two ahead passed over, the ask not yet", seen(ahead(2, ""), taken(delivery.AskDelivery, put.Add(-time.Second))), "", 0},
+			// Note 62: seven messages SEAT_NODE_BUILD_GO gave up on sat ahead of a Go ask for five hours.
+			{"seven given up on ahead, a later ask taken", seen(ahead(7, ""), later), "behind 7 given up on or taken", 0},
+			{"seven taken ahead (failed as poison), a later ask taken", seen(ahead(7, delivery.StepFailed), later), "no longer delivers it", 0},
+			{"a check ask taken later proves a delivery ask passed", seen(ahead(7, ""), taken(delivery.AskCheck, put.Add(time.Minute))), "no longer delivers it", 0},
+			{"a delivery ask taken later does not prove a check ask passed", func() AskSeen { s := seen(ahead(3, ""), later); s.Kind = delivery.AskCheck; return s }(), "", 3},
 		} {
 			got := c.seen.Why()
-			if (c.want == "") != (got == "") || !strings.Contains(got, c.want) {
+			if (c.want == "") != (got == "") || !strings.Contains(got, c.want) || c.seen.Waiting() != c.waiting {
-				t.Errorf("%+v: why %q, want %q", c.seen, got, c.want)
+				t.Errorf("%s: why %q and %d waiting, want %q and %d", c.name, got, c.seen.Waiting(), c.want, c.waiting)
 			}
 		}
+		if at := time.UnixMilli(1793120000123); !AskedAt(strconv.FormatInt(at.UnixMilli(), 36)+"0a1b2c").Equal(at) || !AskedAt("x").IsZero() {
+			t.Errorf("AskedAt does not read an ask id's time")
+		}
 	})
 	t.Run("the tool builder_stuck", func(t *testing.T) { // 1945
 		at := time.Date(2026, 10, 25, 17, 6, 0, 0, time.UTC)

modules/mesh-delivery-builder/manifest_test.go

@@ -87,7 +87,7 @@ func TestManifestDeclarations(t *testing.T) { // 1607/AC1
 	if m.Module != "mesh-delivery-builder" {
 		t.Errorf("module = %q", m.Module)
 	}
-	if want := []state{{Name: "builds", History: 5}, {Name: "instances", History: 1}, {Name: "reasks", History: 1}}; !reflect.DeepEqual(m.State, want) { // delivery.BuilderBuildsHistory
+	if want := []state{{Name: "builds", History: 5}, {Name: "instances", History: 1}, {Name: "queues", History: 1}, {Name: "reasks", History: 1}}; !reflect.DeepEqual(m.State, want) { // delivery.BuilderBuildsHistory
 		t.Errorf("state = %+v, want %+v", m.State, want)
 	}
 	// 1887/A4c: the builder chooses no node, so it reads neither presence nor have, invokes no build module's tool and
@@ -115,8 +115,8 @@ func TestManifestDeclarations(t *testing.T) { // 1607/AC1
 	if m.SecretsOwner != "mesh-delivery-builder" {
 		t.Errorf("secrets-owner = %q", m.SecretsOwner)
 	}
-	// 1945: the builder's tools, served by its tools bundle, and the event it says when it asks an ask again.
+	// 1945: the builder's tools (builder_seat_queue: note 62), served by its tools bundle, and the event it says when it asks an ask again.
-	if want := []string{"builder_ask_again", "builder_stuck"}; !reflect.DeepEqual(sorted(m.Tools), want) {
+	if want := []string{"builder_ask_again", "builder_seat_queue", "builder_stuck"}; !reflect.DeepEqual(sorted(m.Tools), want) {
 		t.Errorf("tools = %v, want %v", m.Tools, want)
 	}
 	if want := []string{"build.asked-again"}; !reflect.DeepEqual(m.Emits, want) {

modules/mesh-delivery-builder/module.json

@@ -4,6 +4,7 @@
   "slug": "builder",
   "tools": [
     "builder_ask_again",
+    "builder_seat_queue",
     "builder_stuck"
   ],
   "emits": [
@@ -20,6 +21,7 @@
   "state": [
     { "name": "builds", "history": 5 },
     { "name": "instances", "history": 1 },
+    { "name": "queues", "history": 1 },
     { "name": "reasks", "history": 1 }
   ],
   "reads": [