Remove only a move's proven second copy, and never delete a handed-out original (hq issue 558)

Second review of #276: a dead letter delivered again beside its original
shares its request id and was removed as a copy, losing the ask; and a
publish that outlived the pause could delete an original a holder had
taken. A copy now names the message it copies, removals are said, the
publish waits inside the pause's margin, and an original handed out
meanwhile keeps its place while the copy goes.

internal/link/queue_order.go

@@ -5,6 +5,7 @@ import (
 	"errors"
 	"fmt"
 	"sort"
+	"strconv"
 	"strings"
 	"sync"
 	"time"
@@ -111,14 +112,26 @@ var orderPasses sync.Mutex
 // 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
+	pausedFor    = 20 * time.Second
+	moveMargin   = 5 * time.Second
+	publishWaits = 3 * 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"
 
+// MovedFromHeader names, on a move's copy, the sequence of the message it is a copy of: with QueuedAtHeader,
+// what proves a second message of one id is a move's copy and not an ask delivered again (ADR 0293), whose
+// headers are its original's — and so never name that original's own sequence.
+const MovedFromHeader = "Mesh-Moved-From"
+
+// Pass is what one pass did: the asks it moved, and each second copy it removed, in words.
+type Pass struct {
+	Moved   []string
+	Removed []string
+}
+
 // 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)
@@ -134,23 +147,24 @@ func signalAsked() {
 // 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. 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) {
+func PutInOrder(ctx context.Context, js *broker.JetStream, seat string) (Pass, error) {
+	var pass Pass
 	orderPasses.Lock()
 	defer orderPasses.Unlock()
 	api, err := jetstream.New(js.Conn())
 	if err != nil {
-		return nil, err
+		return pass, err
 	}
 	worker, _ := broker.HolderConsumerFor("", "", broker.DeclaredSeat{Name: seat, Accepts: []string{"build"}})
 	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
+		return pass, 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, pauseEnds); err != nil {
-			return nil, fmt.Errorf("the %s worker could not be paused to put its queue in order, so it is left "+
+			return pass, 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)
 		}
 		defer func() {
@@ -162,20 +176,21 @@ func PutInOrder(ctx context.Context, js *broker.JetStream, seat string) ([]strin
 			}
 		}()
 		if waiting, extra, _, err = readWaiting(ctx, api, worker, seat); err != nil {
-			return nil, err
+			return pass, err
 		}
 	}
 	stream, err := api.Stream(ctx, worker.Stream)
 	if err != nil {
-		return nil, err
+		return pass, 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)
+			return pass, fmt.Errorf("a second copy of %s (seq %d) could not be removed: %w", a.ID, a.Seq, err)
 		}
+		pass.Removed = append(pass.Removed, fmt.Sprintf("%s at seq %d, a move's second copy of the message at seq %s",
+			a.ID, a.Seq, a.movedFrom))
 	}
-	var moved []string
 	for _, a := range toMove(waiting, QueueOrder(waiting, time.Now())) {
 		if delivered >= 0 && time.Until(pauseEnds) < moveMargin {
 			// The rest waits for the next pass, under a pause of its own.
@@ -183,20 +198,25 @@ func PutInOrder(ctx context.Context, js *broker.JetStream, seat string) ([]strin
 		}
 		ok, err := moveToTheTail(ctx, js, stream, worker.Name, seat, a)
 		if err != nil {
-			return moved, err
+			return pass, err
 		}
 		if ok {
-			moved = append(moved, a.ID)
+			pass.Moved = append(pass.Moved, a.ID)
 		}
 	}
-	return moved, nil
+	return pass, nil
 }
 
 // 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.
+// worker, so nothing is handed out to pause.
+//
+// **A second copy is proven, never guessed from an id**: a waiting message whose MovedFromHeader names an
+// earlier message of the same request id still in the queue, and whose QueuedAtHeader is that message's
+// first-queued time — a move that published its copy and stopped before it deleted the original. The
+// original stays, waiting or handed out (a holder's then, and the copy would build it twice). Any other
+// second message of one id — an ask a dead letter delivered again beside its original (ADR 0293) — is a
+// real ask and stays.
 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)
@@ -214,7 +234,12 @@ func readWaiting(ctx context.Context, api jetstream.JetStream, worker broker.Con
 		return nil, nil, 0, fmt.Errorf("the worker of %s cannot be read: %w", seat, err)
 	}
 	// From the first: the stream holds only what is unsettled, so this is a handful of reads.
-	seen := map[string]bool{}
+	// Every message by sequence, with its id and first-queued time, to prove a copy against.
+	type held struct {
+		id     string
+		stored time.Time
+	}
+	bySeq := map[string]held{}
 	var seq uint64 = 1
 	for n := 0; n < 10000; n++ {
 		msg, err := stream.GetMsg(ctx, seq, jetstream.WithGetMsgSubject(BuildWorkOf(seat)))
@@ -226,10 +251,10 @@ func readWaiting(ctx context.Context, api jetstream.JetStream, worker broker.Con
 		}
 		seq = msg.Sequence + 1
 		a := askOf(msg)
-		again := a.ID != "" && seen[a.ID]
-		if a.ID != "" {
-			seen[a.ID] = true
-		}
+		a.movedFrom = msg.Header.Get(MovedFromHeader)
+		from, known := bySeq[a.movedFrom]
+		again := a.ID != "" && a.movedFrom != "" && known && from.id == a.ID && from.stored.Equal(a.Stored)
+		bySeq[strconv.FormatUint(a.Seq, 10)] = held{id: a.ID, stored: a.Stored}
 		switch {
 		case int64(a.Seq) <= delivered:
 		case again:
@@ -271,14 +296,36 @@ func moveToTheTail(ctx context.Context, js *broker.JetStream, stream jetstream.S
 			again.Header[k] = v
 		}
 	}
-	// Its place in time goes with it: when the stream first took it, as askOf read it.
+	// Its place in time goes with it: when the stream first took it, as askOf read it; and the message it
+	// is a copy of, which proves it one.
 	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))
+	again.Header.Set(MovedFromHeader, strconv.FormatUint(a.Seq, 10))
+	// Inside the margin, so the publish ends while the worker is still paused.
+	publishing, cancel := context.WithTimeout(ctx, publishWaits)
+	ack, 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)
 	}
+	// **Handed out meanwhile** — a bus slow enough that the pause ended: the original is a holder's, and the
+	// copy goes instead, so it is built once.
+	if consumer, err := stream.Consumer(ctx, workerName); err == nil {
+		info, err := consumer.Info(ctx)
+		if err != nil || info.Delivered.Stream >= a.Seq {
+			if derr := stream.DeleteMsg(ctx, ack.Sequence); derr != nil && !errors.Is(derr, jetstream.ErrMsgNotFound) {
+				return false, fmt.Errorf("%s was handed out while it moved and its copy (seq %d) could not be "+
+					"removed; the next pass removes it: %w", a.ID, ack.Sequence, derr)
+			}
+			if err != nil {
+				return false, fmt.Errorf("whether %s was handed out while it moved cannot be read, so its copy was "+
+					"removed and it stays where it was: %w", a.ID, err)
+			}
+			if info.Delivered.Stream >= ack.Sequence {
+				return false, fmt.Errorf("%s was handed out while it moved, and its copy too: it may be built twice", a.ID)
+			}
+			return false, nil
+		}
+	}
 	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)
@@ -319,10 +366,13 @@ func KeepingInOrder(ctx context.Context, js *broker.JetStream, seat func() strin
 		}
 		lastSeq, lastPass = info.State.LastSeq, time.Now()
 		passing, cancel := context.WithTimeout(ctx, pausedFor+10*time.Second)
-		moved, err := PutInOrder(passing, js, name)
+		pass, err := PutInOrder(passing, js, name)
 		cancel()
-		if len(moved) > 0 {
-			said("the %s queue put in order: %s moved behind what goes first", name, strings.Join(moved, ", "))
+		for _, r := range pass.Removed {
+			said("the %s queue: removed %s", name, r)
+		}
+		if len(pass.Moved) > 0 {
+			said("the %s queue put in order: %s moved behind what goes first", name, strings.Join(pass.Moved, ", "))
 		}
 		if err != nil {
 			said("the %s queue could not be put in order: %v", name, err)