From fc573f951a4731ac3b8d1c4fcd03b29058cef787 Mon Sep 17 00:00:00 2001 From: jochen Date: Mon, 12 Oct 2026 21:46:49 +0200 Subject: [PATCH] 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.go | 2 + internal/link/queue_order.go | 104 ++++++++++++++++++++++-------- internal/link/queue_order_test.go | 93 +++++++++++++++++++++++--- 3 files changed, 164 insertions(+), 35 deletions(-) diff --git a/internal/link/queue.go b/internal/link/queue.go index a7c87652..aef98d3a 100644 --- a/internal/link/queue.go +++ b/internal/link/queue.go @@ -65,6 +65,8 @@ type QueuedAsk struct { // Stored is when the stream first took it, a move into order kept (novox/hq issue 558). Stored time.Time `json:"stored,omitempty"` State string `json:"state"` + // movedFrom is the MovedFromHeader it carries: the sequence of the message a move copied it from. + movedFrom string // On and Started are the holder building an in-flight ask and when it started, from its started // event. Was is the holder that started an in-flight ask and is no longer building it — handed // back, to be handed out again — with Started its start there. diff --git a/internal/link/queue_order.go b/internal/link/queue_order.go index c776dc42..449618f9 100644 --- a/internal/link/queue_order.go +++ b/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) diff --git a/internal/link/queue_order_test.go b/internal/link/queue_order_test.go index efcf9284..315b7dac 100644 --- a/internal/link/queue_order_test.go +++ b/internal/link/queue_order_test.go @@ -200,9 +200,9 @@ func TestNatsAMoveCutShortLeavesOneCopyAfterTheNextPass(t *testing.T) { check := aCheck(now.Add(-2*time.Minute), "/c") publishAsk(t, js, check) publishAsk(t, js, aDelivery(now.Add(-time.Minute), "/d")) - publishAsk(t, js, check) // the copy, published; the original never deleted - if _, err := PutInOrder(ctx, js, TheBuildMachine); err != nil { - t.Fatal(err) + moveCopyOf(t, js, check.ID) // the copy, published; the original never deleted + if pass, err := PutInOrder(ctx, js, TheBuildMachine); err != nil || len(pass.Removed) != 1 { + t.Fatalf("removed %v: %v", pass.Removed, err) } var got []string for _, a := range queued(t, js) { @@ -234,7 +234,7 @@ func TestNatsAMoveCutShortLeavesOneCopyAfterTheNextPass(t *testing.T) { } _ = msgs[0].Ack() } - publishAsk(t, js, taken) + moveCopyOf(t, js, taken.ID) if _, err := PutInOrder(ctx, js, TheBuildMachine); err != nil { t.Fatal(err) } @@ -249,6 +249,83 @@ func TestNatsAMoveCutShortLeavesOneCopyAfterTheNextPass(t *testing.T) { } } +// moveCopyOf publishes what a move publishes for the ask with this id, as a pass stopped before its delete +// would leave it. +func moveCopyOf(t *testing.T, js *broker.JetStream, id string) { + t.Helper() + for _, a := range queued(t, js) { + if a.ID != id { + continue + } + body, _ := json.Marshal(a.Request) + msg := &nats.Msg{Subject: BuildWork(), Data: body, Header: nats.Header{}} + msg.Header.Set(QueuedAtHeader, a.Stored.UTC().Format(time.RFC3339Nano)) + msg.Header.Set(MovedFromHeader, fmt.Sprint(a.Seq)) + if _, err := js.Context().PublishMsg(msg); err != nil { + t.Fatal(err) + } + return + } + t.Fatalf("%s is not queued", id) +} + +// A dead letter delivered again whose original stayed beside it (ADR 0293: its removal failed) is a real +// ask with the same id, and is never removed as a second copy — even when the original was itself a move's +// copy, whose headers the delivery carries. +func TestNatsAnAskDeliveredAgainBesideItsOriginalStays(t *testing.T) { + js := aBusWithTheBuildRole(t) + ctx := t.Context() + ask := aCheck(time.Now(), "/again") + publishAsk(t, js, aCheck(time.Now().Add(-time.Minute), "/before")) + publishAsk(t, js, ask) + publishAsk(t, js, aDelivery(time.Now(), "/d")) + if _, err := PutInOrder(ctx, js, TheBuildMachine); err != nil { // /again becomes a move's copy + t.Fatal(err) + } + worker, _ := broker.HolderConsumerFor("", "", broker.DeclaredSeat{Name: TheBuildMachine, Accepts: []string{"build"}}) + sub, err := js.Context().PullSubscribe(worker.Filters[0], worker.Name, nats.Bind(worker.Stream, worker.Name), nats.ManualAck()) + if err != nil { + t.Fatal(err) + } + defer func() { _ = sub.Unsubscribe() }() + var original *nats.Msg + for original == nil { + msgs, err := sub.Fetch(1, nats.MaxWait(2*time.Second)) + if err != nil || len(msgs) == 0 { + t.Fatalf("nothing to take: %v", err) + } + var r BuildRequest + _ = json.Unmarshal(msgs[0].Data, &r) + if r.ID == ask.ID { + original = msgs[0] // given up on, never settled: dead, and kept + continue + } + _ = msgs[0].Ack() + } + // Delivered again as DeliverAgain does: its headers, and its own. + again := &nats.Msg{Subject: BuildWork(), Data: original.Data, Header: nats.Header{}} + for k, v := range original.Header { + again.Header[k] = v + } + again.Header.Set(AgainHeader, "7") + if _, err := js.Context().PublishMsg(again); err != nil { + t.Fatal(err) + } + pass, err := PutInOrder(ctx, js, TheBuildMachine) + if err != nil || len(pass.Removed) != 0 { + t.Fatalf("removed %v: %v", pass.Removed, err) + } + n := 0 + for _, a := range queued(t, js) { + if a.ID == ask.ID { + n++ + } + } + if n != 2 { + t.Fatalf("%d messages of the ask delivered again are queued, want its original and the delivery", n) + } +} + // Two passes at once: nothing is built twice, and nothing is lost. func TestNatsTwoPassesAtOnceBuildNothingTwice(t *testing.T) { js := aBusWithTheBuildRole(t) @@ -327,8 +404,8 @@ func TestNatsACancelRacingAMoveDeletesTheMovedCopy(t *testing.T) { publishAsk(t, js, aCheck(now.Add(-time.Minute), "/c")) read := queued(t, js)[0] publishAsk(t, js, aDelivery(now, "/d")) - if moved, err := PutInOrder(ctx, js, TheBuildMachine); err != nil || len(moved) != 1 { - t.Fatalf("moved %v: %v", moved, err) + if pass, err := PutInOrder(ctx, js, TheBuildMachine); err != nil || len(pass.Moved) != 1 { + t.Fatalf("moved %v: %v", pass.Moved, err) } removed, err := DeleteAsk(ctx, js, TheBuildMachine, read) if err != nil || removed == 0 || removed == read.Seq { @@ -354,8 +431,8 @@ func TestNatsAnUnreadableAskKeepsItsAgeWhenItMoves(t *testing.T) { first := queued(t, js)[0].Stored time.Sleep(50 * time.Millisecond) publishAsk(t, js, aDelivery(time.Now(), "/d")) - if moved, err := PutInOrder(ctx, js, TheBuildMachine); err != nil || len(moved) != 1 { - t.Fatalf("moved %v: %v", moved, err) + if pass, err := PutInOrder(ctx, js, TheBuildMachine); err != nil || len(pass.Moved) != 1 { + t.Fatalf("moved %v: %v", pass.Moved, err) } asks := queued(t, js) if len(asks) != 2 || asks[1].ID != "" || !asks[1].Stored.Equal(first) {