Builder: count only deliverable asks ahead, drop the old message of an ask asked again, and read a seat's queue (issue 1945)

mesh-delivery #51's re-ask never fired for mesh-build-node.4e7b118e: messages the
SEAT_NODE_BUILD_GO workers had given up on stay on the work queue, and each counted
as waiting ahead, so the ask was never asked again (note 62). A message ahead now
waits only when no node wrote its step and no ask of its kind asked after it was
taken; the old message of an ask asked again is deleted from its stream; and
builder_seat_queue (answered by the builder through its queues state, as only its
account may read the stream) shows each message and whether it still waits.

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.
-func Tools(r build.StuckReader, askAgain func(key string, value []byte) error, now func() time.Time) []stdio.Tool {
+// queueAsker writes a queues value and waits for the builder's answer to it.
+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,
-// when the ask waits first in line.
+// maxOnSubject bounds the messages of one subject a read walks (each a read by sequence, next by subject), and maxSteps
+// 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)
-	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
-	}
+	stream, name, subject, err := h.stream(ctx, a.Language, a.Kind)
 	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
-	// instead, each a direct read that consumes nothing.
-	info, err := stream.Info(ctx)
+	var ahead []build.Queued
+	err = onSubject(ctx, stream, subject, func(q build.Queued) bool {
+		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++ {
-		m, err := stream.GetMsg(ctx, seq)
+	for i := range ahead {
+		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 {
-			continue
+		q := build.Queued{Seq: m.Sequence, Ask: m.Header.Get(jetstream.MsgIDHeader), PutAt: m.Time}
+		if _, kind, err := delivery.ParseAskSubject(m.Subject); err == nil {
+			q.Kind = kind
 		}
-		if m.Header.Get(jetstream.MsgIDHeader) == a.ID {
-			seen.OnStream, seen.PutAt = true, m.Time
-			break
+		if q.At = build.AskedAt(q.Ask); q.At.IsZero() {
+			q.At = q.PutAt
 		}
-		seen.Ahead++
+		if !each(q) {
+			return nil
+		}
+		seq = m.Sequence + 1
 	}
-	if !later || !seen.OnStream || seen.Ahead > 0 {
-		return seen, nil
-	}
-	// First in line and not taken: did a build node take any ask put after it? Its stream then no longer delivers it.
+}
+
+// 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
+// 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 err != nil || !s.Started.After(seen.PutAt) {
-			continue // not a step (a node's pause key), or one started before the ask was put
+		if s, err := delivery.DecodeStep(e.Value()); err == nil {
+			return s.State
 		}
-		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)
+	}
+}