Make a queue move lossless and run its passes on the acting controller alone (hq issue 558)

Review of #276: a move deleted before it published, so a failed publish
lost an ask; a verb's own process ran a pass beside the serving one, so
one could resume the queue while the other moved asks and a build ran
twice. The copy is now published first and second copies removed by id,
asks only signal the acting controller's loop, moves stop with 5 s of
the pause left, a cancel finds a moved ask by id, and an unreadable ask
keeps its age.

internal/link/queue_order.go

@@ -103,15 +103,37 @@ func toMove(queue, order []QueuedAsk) []QueuedAsk {
 	return order[kept:]
 }
 
-// orderPasses keeps one pass at a time in this process: two would each resume the worker the other paused.
+// orderPasses keeps one pass at a time. Only the acting controller's loop runs one (KeepingInOrder), so
+// there is one; the lock holds a test's two at once to that too.
 var orderPasses sync.Mutex
 
 // pausedFor bounds a pause the pass could not end itself: a controller that stops mid-pass leaves the
-// worker paused this long at most.
-const pausedFor = 20 * time.Second
+// worker paused this long at most. moveMargin is the part of it no move starts in: a pass stops moving
+// with at least this left, and the next pass moves the rest, so no move runs past the pause.
+const (
+	pausedFor  = 20 * time.Second
+	moveMargin = 5 * time.Second
+)
+
+// QueuedAtHeader carries, on a moved ask, when the stream first took it: a move keeps an ask's place in
+// time, its id unread or not.
+const QueuedAtHeader = "Mesh-Queued-At"
+
+// asked is the signal an ask sends the acting controller's pass loop, in this process. An ask asked in
+// another process — a verb run as its own command — is seen by the loop as the queue's last sequence moving.
+var asked = make(chan struct{}, 1)
+
+// signalAsked tells the pass loop an ask joined the queue. Never blocks, never runs a pass.
+func signalAsked() {
+	select {
+	case asked <- struct{}{}:
+	default:
+	}
+}
 
 // PutInOrder moves the seat's waiting asks into the order QueueOrder says, and says each it moved. A queue
-// already in order is read and left alone; the worker is paused only while asks move.
+// already in order is read and left alone; the worker is paused only while asks move. Run by the acting
+// controller alone (KeepingInOrder): two passes in two processes would each resume what the other paused.
 func PutInOrder(ctx context.Context, js *broker.JetStream, seat string) ([]string, error) {
 	orderPasses.Lock()
 	defer orderPasses.Unlock()
@@ -120,13 +142,14 @@ func PutInOrder(ctx context.Context, js *broker.JetStream, seat string) ([]strin
 		return nil, err
 	}
 	worker, _ := broker.HolderConsumerFor("", "", broker.DeclaredSeat{Name: seat, Accepts: []string{"build"}})
-	waiting, delivered, err := readWaiting(ctx, api, worker, seat)
-	if err != nil || len(toMove(waiting, QueueOrder(waiting, time.Now()))) == 0 {
+	waiting, extra, delivered, err := readWaiting(ctx, api, worker, seat)
+	if err != nil || (len(extra) == 0 && len(toMove(waiting, QueueOrder(waiting, time.Now()))) == 0) {
 		return nil, err
 	}
 	// Paused, then read again: what the worker handed out up to the pause is not waiting any more.
+	pauseEnds := time.Now().Add(pausedFor)
 	if delivered >= 0 {
-		if _, err := api.PauseConsumer(ctx, worker.Stream, worker.Name, time.Now().Add(pausedFor)); err != nil {
+		if _, err := api.PauseConsumer(ctx, worker.Stream, worker.Name, pauseEnds); err != nil {
 			return nil, fmt.Errorf("the %s worker could not be paused to put its queue in order, so it is left "+
 				"first come, first served: %w", seat, err)
 		}
@@ -135,16 +158,30 @@ func PutInOrder(ctx context.Context, js *broker.JetStream, seat string) ([]strin
 			defer cancel()
 			if _, err := api.ResumeConsumer(resuming, worker.Stream, worker.Name); err != nil {
 				fmt.Printf("the %s worker could not be resumed after its queue was put in order; it resumes by "+
-					"itself within %s: %v\n", seat, pausedFor, err)
+					"itself by %s: %v\n", seat, pauseEnds.Format(time.RFC3339), err)
 			}
 		}()
-		if waiting, _, err = readWaiting(ctx, api, worker, seat); err != nil {
+		if waiting, extra, _, err = readWaiting(ctx, api, worker, seat); err != nil {
 			return nil, err
 		}
 	}
+	stream, err := api.Stream(ctx, worker.Stream)
+	if err != nil {
+		return nil, err
+	}
+	// A second copy of one ask — a move that published and stopped before it deleted — goes first.
+	for _, a := range extra {
+		if err := stream.DeleteMsg(ctx, a.Seq); err != nil && !errors.Is(err, jetstream.ErrMsgNotFound) {
+			return nil, fmt.Errorf("a second copy of %s (seq %d) could not be removed: %w", a.ID, a.Seq, err)
+		}
+	}
 	var moved []string
 	for _, a := range toMove(waiting, QueueOrder(waiting, time.Now())) {
-		ok, err := moveToTheTail(ctx, js, api, worker, seat, a.Seq)
+		if delivered >= 0 && time.Until(pauseEnds) < moveMargin {
+			// The rest waits for the next pass, under a pause of its own.
+			break
+		}
+		ok, err := moveToTheTail(ctx, js, stream, worker.Name, seat, a)
 		if err != nil {
 			return moved, err
 		}
@@ -155,71 +192,78 @@ func PutInOrder(ctx context.Context, js *broker.JetStream, seat string) ([]strin
 	return moved, nil
 }
 
-// readWaiting is every ask the worker has not yet handed out, in queue order, and the last sequence it
-// handed out; -1 when there is no worker, so nothing is handed out to pause.
-func readWaiting(ctx context.Context, api jetstream.JetStream, worker broker.Consumer, seat string) ([]QueuedAsk, int64, error) {
+// readWaiting is every ask the worker has not yet handed out, in queue order, but each second copy of one
+// ask; the second copies, to remove; and the last sequence the worker handed out, -1 when there is no
+// worker, so nothing is handed out to pause. A second copy is a waiting ask whose request id another ask
+// before it carries, or one the worker already handed out: that one is a holder's, and the copy would build
+// it twice.
+func readWaiting(ctx context.Context, api jetstream.JetStream, worker broker.Consumer, seat string) (
+	waiting, extra []QueuedAsk, delivered int64, err error) {
 	stream, err := api.Stream(ctx, worker.Stream)
 	if err != nil {
-		return nil, 0, fmt.Errorf("the %s work queue (%s) cannot be read: %w", seat, worker.Stream, err)
+		return nil, nil, 0, fmt.Errorf("the %s work queue (%s) cannot be read: %w", seat, worker.Stream, err)
 	}
-	delivered := int64(-1)
+	delivered = -1
 	if consumer, err := stream.Consumer(ctx, worker.Name); err == nil {
 		info, err := consumer.Info(ctx)
 		if err != nil {
-			return nil, 0, fmt.Errorf("the worker of %s cannot be read: %w", seat, err)
+			return nil, nil, 0, fmt.Errorf("the worker of %s cannot be read: %w", seat, err)
 		}
 		delivered = int64(info.Delivered.Stream)
 	} else if !errors.Is(err, jetstream.ErrConsumerNotFound) {
-		return nil, 0, fmt.Errorf("the worker of %s cannot be read: %w", seat, err)
+		return nil, nil, 0, fmt.Errorf("the worker of %s cannot be read: %w", seat, err)
 	}
-	var out []QueuedAsk
-	seq := uint64(max(delivered, 0)) + 1
+	// From the first: the stream holds only what is unsettled, so this is a handful of reads.
+	seen := map[string]bool{}
+	var seq uint64 = 1
 	for n := 0; n < 10000; n++ {
 		msg, err := stream.GetMsg(ctx, seq, jetstream.WithGetMsgSubject(BuildWorkOf(seat)))
 		if errors.Is(err, jetstream.ErrMsgNotFound) {
 			break
 		}
 		if err != nil {
-			return nil, 0, fmt.Errorf("reading the %s work queue: %w", seat, err)
+			return nil, nil, 0, fmt.Errorf("reading the %s work queue: %w", seat, err)
 		}
-		out = append(out, askOf(msg))
 		seq = msg.Sequence + 1
+		a := askOf(msg)
+		again := a.ID != "" && seen[a.ID]
+		if a.ID != "" {
+			seen[a.ID] = true
+		}
+		switch {
+		case int64(a.Seq) <= delivered:
+		case again:
+			extra = append(extra, a)
+		default:
+			waiting = append(waiting, a)
+		}
 	}
-	return out, delivered, nil
+	return waiting, extra, delivered, nil
 }
 
-// moveToTheTail deletes the ask at seq and publishes its body again at the queue's tail. Deleted first, as
-// a dead letter delivered again is (novox/hq ADR 0293): one already gone — taken, cancelled, or moved by
-// another pass — is not published twice. And looked at once more just before: one the worker has handed
-// out since — another process's pass resumed it — is a holder's now and stays. False when it was not moved.
-func moveToTheTail(ctx context.Context, js *broker.JetStream, api jetstream.JetStream, worker broker.Consumer, seat string,
-	seq uint64) (bool, error) {
-	stream, err := api.Stream(ctx, worker.Stream)
-	if err != nil {
-		return false, err
-	}
-	if consumer, err := stream.Consumer(ctx, worker.Name); err == nil {
+// moveToTheTail publishes the ask's body again at the queue's tail and then deletes it where it was. **The
+// copy first**: a publish that fails deletes nothing, and a pass that stops between the two leaves two
+// copies, which the next pass sees by their one id and makes one (readWaiting). A move never loses an ask.
+// Looked at once more just before: one the worker has handed out since is a holder's now and stays. False
+// when it was not moved.
+func moveToTheTail(ctx context.Context, js *broker.JetStream, stream jetstream.Stream, workerName, seat string,
+	a QueuedAsk) (bool, error) {
+	if consumer, err := stream.Consumer(ctx, workerName); err == nil {
 		info, err := consumer.Info(ctx)
 		if err != nil {
 			return false, err
 		}
-		if info.Delivered.Stream >= seq {
+		if info.Delivered.Stream >= a.Seq {
 			return false, nil
 		}
 	}
-	msg, err := stream.GetMsg(ctx, seq)
+	msg, err := stream.GetMsg(ctx, a.Seq)
 	if errors.Is(err, jetstream.ErrMsgNotFound) {
 		return false, nil
 	}
 	if err != nil {
 		return false, err
 	}
-	if err := stream.DeleteMsg(ctx, seq); err != nil {
-		if errors.Is(err, jetstream.ErrMsgNotFound) || errors.Is(err, jetstream.ErrMsgDeleteUnsuccessful) {
-			return false, nil
-		}
-		return false, err
-	}
 	again := &nats.Msg{Subject: msg.Subject, Data: msg.Data, Header: nats.Header{}}
 	for k, v := range msg.Header {
 		// The server's own headers stay behind: a message id would be taken for a duplicate and dropped.
@@ -227,17 +271,106 @@ func moveToTheTail(ctx context.Context, js *broker.JetStream, api jetstream.JetS
 			again.Header[k] = v
 		}
 	}
-	for try := 0; ; try++ {
-		publishing, cancel := context.WithTimeout(ctx, 10*time.Second)
-		_, err := js.Context().PublishMsg(again, nats.Context(publishing))
+	// Its place in time goes with it: when the stream first took it, as askOf read it.
+	again.Header.Set(QueuedAtHeader, askOf(msg).Stored.UTC().Format(time.RFC3339Nano))
+	publishing, cancel := context.WithTimeout(ctx, 10*time.Second)
+	_, err = js.Context().PublishMsg(again, nats.Context(publishing))
+	cancel()
+	if err != nil {
+		return false, fmt.Errorf("%s (seq %d) could not be put behind, and stays where it is: %w", a.ID, a.Seq, err)
+	}
+	if err := stream.DeleteMsg(ctx, a.Seq); err != nil && !errors.Is(err, jetstream.ErrMsgNotFound) {
+		return false, fmt.Errorf("%s was copied behind and its original (seq %d) not removed; the next pass "+
+			"removes the second copy: %w", a.ID, a.Seq, err)
+	}
+	return true, nil
+}
+
+// KeepingInOrder runs the passes, on the acting controller alone, until ctx ends: when an ask in this
+// process says so, when the queue's last sequence moved — an ask from another process — and every
+// passEvery, for a check that comes of age with nothing asked. seat says the build seat to order; acting
+// whether this controller acts (Holding); said what a pass did.
+func KeepingInOrder(ctx context.Context, js *broker.JetStream, seat func() string, acting func() bool,
+	said func(format string, args ...any)) {
+	tick := time.NewTicker(orderLooks)
+	defer tick.Stop()
+	var lastSeq uint64
+	var lastPass time.Time
+	for {
+		signalled := false
+		select {
+		case <-ctx.Done():
+			return
+		case <-asked:
+			signalled = true
+		case <-tick.C:
+		}
+		if !acting() {
+			continue
+		}
+		name := seat()
+		worker, _ := broker.HolderConsumerFor("", "", broker.DeclaredSeat{Name: name, Accepts: []string{"build"}})
+		info, err := js.Context().StreamInfo(worker.Stream)
+		if err != nil {
+			continue
+		}
+		if !signalled && info.State.LastSeq == lastSeq && time.Since(lastPass) < passEvery {
+			continue
+		}
+		lastSeq, lastPass = info.State.LastSeq, time.Now()
+		passing, cancel := context.WithTimeout(ctx, pausedFor+10*time.Second)
+		moved, err := PutInOrder(passing, js, name)
 		cancel()
-		if err == nil {
-			return true, nil
+		if len(moved) > 0 {
+			said("the %s queue put in order: %s moved behind what goes first", name, strings.Join(moved, ", "))
 		}
-		if try == 2 {
-			return false, fmt.Errorf("the ask at seq %d of the %s queue was taken out to be put behind and could "+
-				"NOT be put back: %w — its asker hears no outcome and must ask again", seq, seat, err)
+		if err != nil {
+			said("the %s queue could not be put in order: %v", name, err)
+		}
+		if info, err := js.Context().StreamInfo(worker.Stream); err == nil {
+			// Its own moves are not an ask.
+			lastSeq = info.State.LastSeq
 		}
-		time.Sleep(time.Second)
 	}
 }
+
+// orderLooks is how often the loop looks whether an ask joined from another process; passEvery how often
+// it passes with nothing asked.
+var (
+	orderLooks = 2 * time.Second
+	passEvery  = 30 * time.Second
+)
+
+// DeleteAsk removes a waiting or dead ask from the queue: at its sequence, or — moved into order since it
+// was read — the copy carrying its id. Says the sequence it removed, 0 when neither was there any more.
+func DeleteAsk(ctx context.Context, js *broker.JetStream, seat string, ask QueuedAsk) (uint64, error) {
+	worker, _ := broker.HolderConsumerFor("", "", broker.DeclaredSeat{Name: seat, Accepts: []string{"build"}})
+	gone := func(err error) bool {
+		return errors.Is(err, nats.ErrMsgNotFound) || errors.Is(err, jetstream.ErrMsgNotFound) ||
+			(err != nil && strings.Contains(err.Error(), "no message found"))
+	}
+	err := js.Context().DeleteMsg(worker.Stream, ask.Seq)
+	if err == nil {
+		return ask.Seq, nil
+	}
+	if !gone(err) {
+		return 0, err
+	}
+	if ask.ID == "" {
+		return 0, nil
+	}
+	q, err := ReadQueue(ctx, js, seat)
+	if err != nil {
+		return 0, err
+	}
+	for _, a := range q.Asks {
+		if a.ID == ask.ID && a.Seq != ask.Seq && a.State != AskInFlight {
+			if err := js.Context().DeleteMsg(worker.Stream, a.Seq); err != nil && !gone(err) {
+				return 0, err
+			} else if err == nil {
+				return a.Seq, nil
+			}
+		}
+	}
+	return 0, nil
+}