From c6e234455f5e10c1545a41f9f5ca9f75da9a316e Mon Sep 17 00:00:00 2001 From: jochen Date: Mon, 12 Oct 2026 21:25:46 +0200 Subject: [PATCH 1/3] Hand a delivery build out before pull-request checks (hq issue 558) A merged change's walk waited minutes on its build behind checks in a first-come queue. The queue now hands deliveries out first, a check that has waited ten minutes before deliveries asked after it, and says each ask's kind; the order is kept in the work queue itself. --- cmd/mesh-controller/queue.go | 50 ++++- cmd/mesh-controller/queue_answer_test.go | 11 +- cmd/mesh-controller/watchdogs.go | 3 + internal/link/builds_nats.go | 13 ++ internal/link/queue.go | 30 ++- internal/link/queue_order.go | 243 +++++++++++++++++++++++ internal/link/queue_order_test.go | 124 ++++++++++++ 7 files changed, 453 insertions(+), 21 deletions(-) create mode 100644 internal/link/queue_order.go create mode 100644 internal/link/queue_order_test.go diff --git a/cmd/mesh-controller/queue.go b/cmd/mesh-controller/queue.go index 567357fd..9ef17397 100644 --- a/cmd/mesh-controller/queue.go +++ b/cmd/mesh-controller/queue.go @@ -81,6 +81,39 @@ func listQueue(ctx context.Context, args []string, w io.Writer) error { return nil } +// keepingTheQueueInOrder puts the build queue in order every watchEvery while this controller acts +// (novox/hq issue 558). An ask puts it in order as it joins; this is for the clock: a check comes of age +// with nothing asked. +func keepingTheQueueInOrder(ctx context.Context, open *stores, js *broker.JetStream) { + tick := time.NewTicker(watchEvery) + defer tick.Stop() + for { + select { + case <-ctx.Done(): + return + case <-tick.C: + } + if !link.Holding() || js == nil { + continue + } + entries, err := open.inventory.Catalogued(ctx) + if err != nil { + fmt.Printf("the build queue is not put in order: the catalogue cannot be read: %v\n", err) + continue + } + seat := buildSeatAmong(entries) + ordering, cancel := context.WithTimeout(ctx, watchEvery) + moved, err := link.PutInOrder(ordering, js, seat) + cancel() + if len(moved) > 0 { + fmt.Printf("the %s queue put in order: %s moved behind what goes first\n", seat, strings.Join(moved, ", ")) + } + if err != nil { + fmt.Printf("the %s queue could not be put in order: %v\n", seat, err) + } + } +} + // queueAnswer is the queue as data (novox/hq issue 455): what a caller reads to say how far back an ask // waits — "waiting in the build queue behind 2" — which the table alone could not tell it. Text is the // table, written from the same read: the verb answers it as its output and the rest as its answer, so the @@ -100,7 +133,8 @@ type queueAnswer struct { type queuedAnswer struct { ID string `json:"id"` State string `json:"state"` - // Kind is "check" for a pull request's merge check, "build" for anything else. + // Kind is "delivery" for a build a walk waits for, "check" for a pull request's check and any ask + // whose kind cannot be told (link.KindOf, novox/hq issue 558). Kind string `json:"kind"` Repository string `json:"repository"` Path string `json:"path,omitempty"` @@ -129,10 +163,7 @@ func queueAnswerOf(q link.Queue, now time.Time) queueAnswer { out := queueAnswer{Seat: q.Seat, Waiting: len(waiting), InFlight: len(running), Dead: len(dead), MaxDeliver: q.MaxDeliver, Asks: []queuedAnswer{}, Text: queueText(q, now)} one := func(a link.QueuedAsk, position int) queuedAnswer { - kind := "build" - if a.Request.Check != nil { - kind = "check" - } + kind := link.KindOf(a.Request) since := a.AskedAt if a.State == link.AskInFlight && a.On != "" && !a.Started.IsZero() { since = a.Started @@ -189,19 +220,20 @@ func queueText(q link.Queue, now time.Time) string { case a.Was != "": where = fmt.Sprintf("handed back by %s, to be handed out again", a.Was) } - fmt.Fprintf(&b, " %-26s %s — %s, %s (seq %d)\n", a.ID, what(a), where, ago(a.AskedAt), a.Seq) + fmt.Fprintf(&b, " %-26s %-8s %s — %s, %s (seq %d)\n", a.ID, link.KindOf(a.Request), what(a), where, + ago(a.AskedAt), a.Seq) } } if len(waiting) > 0 { - fmt.Fprintln(&b, "\nwaiting:") + fmt.Fprintln(&b, "\nwaiting, in the order they are handed out:") for _, a := range waiting { - fmt.Fprintf(&b, " %-26s %s — %s (seq %d)\n", a.ID, what(a), ago(a.AskedAt), a.Seq) + fmt.Fprintf(&b, " %-26s %-8s %s — %s (seq %d)\n", a.ID, link.KindOf(a.Request), what(a), ago(a.AskedAt), a.Seq) } } if len(dead) > 0 { fmt.Fprintf(&b, "\ndead — handed out as often as the worker allows (%d) and never settled, still held:\n", q.MaxDeliver) for _, a := range dead { - fmt.Fprintf(&b, " %-26s %s — %s (seq %d)\n", a.ID, what(a), ago(a.AskedAt), a.Seq) + fmt.Fprintf(&b, " %-26s %-8s %s — %s (seq %d)\n", a.ID, link.KindOf(a.Request), what(a), ago(a.AskedAt), a.Seq) } } switch { diff --git a/cmd/mesh-controller/queue_answer_test.go b/cmd/mesh-controller/queue_answer_test.go index d56abddd..ca6fed93 100644 --- a/cmd/mesh-controller/queue_answer_test.go +++ b/cmd/mesh-controller/queue_answer_test.go @@ -24,7 +24,7 @@ func aQueueOfEachKind(now time.Time) link.Queue { State: link.AskInFlight, On: "ace", Started: started, Request: link.BuildRequest{ID: "build-running", Check: &link.CheckRequest{Owner: "novox", Repo: "mesh-host"}}}, {Seq: 3, ID: "build-first", Repository: "ssh://forge/novox/mesh-catalog", Ref: "main", AskedAt: asked, - State: link.AskWaiting}, + State: link.AskWaiting, Request: link.BuildRequest{ID: "build-first", Repository: "ssh://forge/novox/mesh-catalog"}}, {Seq: 4, ID: "build-second", Repository: "ssh://forge/novox/mesh-catalog", Path: "modules/gitea", Ref: aCommit, AskedAt: now.Add(-time.Minute), State: link.AskWaiting, Request: link.BuildRequest{ID: "build-second", Check: &link.CheckRequest{Owner: "novox", Repo: "mesh-catalog"}}}, @@ -51,7 +51,7 @@ func TestTheQueueVerbAnswersEachAskAsData(t *testing.T) { t.Errorf("in flight: %+v", running) } first, second := byID["build-first"], byID["build-second"] - if first.Position != 1 || first.Kind != "build" || first.Commit != "" || first.Ref != "main" || first.Machine != "" || + if first.Position != 1 || first.Kind != "delivery" || first.Commit != "" || first.Ref != "main" || first.Machine != "" || !first.Since.Equal(now.Add(-5*time.Minute)) { t.Errorf("first waiting: %+v", first) } @@ -62,6 +62,13 @@ func TestTheQueueVerbAnswersEachAskAsData(t *testing.T) { if dead := byID["build-dead"]; dead.Position != 0 || dead.State != link.AskDead { t.Errorf("dead: %+v", dead) } + // And the table says each ask's kind too (novox/hq issue 558). + for _, line := range []string{"build-running check ", "build-first delivery ", + "build-second check ", "build-dead check "} { + if !strings.Contains(got.Text, line) { + t.Errorf("the table does not say %q:\n%s", line, got.Text) + } + } // In the table's order: in flight, then the waiting by position, then the dead. var order []string for _, a := range got.Asks { diff --git a/cmd/mesh-controller/watchdogs.go b/cmd/mesh-controller/watchdogs.go index fc3fd58c..428744a2 100644 --- a/cmd/mesh-controller/watchdogs.go +++ b/cmd/mesh-controller/watchdogs.go @@ -739,6 +739,9 @@ func watchTheMesh(ctx context.Context, open *stores, server *link.Server, bus li // and the answer chosen is performed on its warrant. startAsking(watching, open, server, bus.Conn, keeper) go forgettingOldHeals(watching, open.inventory) + // And the build queue's order (novox/hq issue 558): a check that has waited its while goes ahead of the + // deliveries asked after it even when nothing new is asked. + go keepingTheQueueInOrder(watching, open, server.JetStream()) fmt.Printf("watching the mesh: %d signal(s) every %s, %d probe(s) every %s; what is wrong is kept in %s "+ "and said as %s events\n", len(watchedRows()), watchEvery, len(runnableProbes()), doctorEvery, broker.ConditionsBucket, conditions.Seat) diff --git a/internal/link/builds_nats.go b/internal/link/builds_nats.go index 1358dd92..3e11fca2 100644 --- a/internal/link/builds_nats.go +++ b/internal/link/builds_nats.go @@ -79,9 +79,21 @@ func (b *natsBuilds) Ask(ctx context.Context, request BuildRequest) error { if _, err := b.js.Context().Publish(BuildWorkOf(b.role()), body, nats.Context(publish)); err != nil { return fmt.Errorf("cannot submit a build: %w", err) } + b.putInOrder(ctx) return nil } +// putInOrder puts the queue in order after an ask joined it (novox/hq issue 558): a delivery goes ahead of +// the checks waiting before it now, not when the serving controller next looks. Never the ask's failure: +// the ask is in the queue, and a queue left first come, first served still answers it. +func (b *natsBuilds) putInOrder(ctx context.Context) { + ordering, cancel := context.WithTimeout(ctx, 15*time.Second) + defer cancel() + if _, err := PutInOrder(ordering, b.js, b.role()); err != nil { + fmt.Fprintf(os.Stderr, "the %s queue could not be put in order: %v\n", b.role(), err) + } +} + // ErrNotHandedOver is a Submit that failed before the build was handed to the build seat: nothing runs. Any // other error from Submit came after: the build may still run, and its outcome may still come (novox/hq // issue 325). @@ -113,6 +125,7 @@ func (b *natsBuilds) Submit(ctx context.Context, request BuildRequest, if _, err := b.js.Context().Publish(BuildWorkOf(b.role()), body, nats.Context(publish)); err != nil { return BuildResult{}, fmt.Errorf("%w: cannot submit a build: %w", ErrNotHandedOver, err) } + b.putInOrder(ctx) waiting, cancelWait := context.WithTimeout(ctx, wait) defer cancelWait() diff --git a/internal/link/queue.go b/internal/link/queue.go index 41abbb9e..b70cc7d3 100644 --- a/internal/link/queue.go +++ b/internal/link/queue.go @@ -62,7 +62,9 @@ type QueuedAsk struct { Path string `json:"path,omitempty"` Ref string `json:"ref,omitempty"` AskedAt time.Time `json:"asked-at,omitempty"` - State string `json:"state"` + // Stored is when the stream took it: when it was asked, or last moved into order (novox/hq issue 558). + Stored time.Time `json:"stored,omitempty"` + State string `json:"state"` // 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. @@ -236,15 +238,7 @@ func ReadQueue(ctx context.Context, js *broker.JetStream, seat string) (Queue, e if err != nil { return q, fmt.Errorf("reading the %s work queue: %w", seat, err) } - ask := QueuedAsk{Seq: msg.Sequence} - if json.Unmarshal(msg.Data, &ask.Request) == nil { - r := ask.Request - ask.ID, ask.Repository, ask.Path, ask.Ref = r.ID, r.Repository, r.Path, r.Ref - if at, ok := BuildAskedAt(r.ID); ok { - ask.AskedAt = at - } - } - asks = append(asks, ask) + asks = append(asks, askOf(msg)) seq = msg.Sequence + 1 } @@ -269,6 +263,22 @@ func ReadQueue(ctx context.Context, js *broker.JetStream, seat string) (Queue, e return q, nil } +// askOf is one ask as the stream holds it: what its request says, and when the stream took it. +func askOf(msg *jetstream.RawStreamMsg) QueuedAsk { + ask := QueuedAsk{Seq: msg.Sequence, Stored: msg.Time} + if json.Unmarshal(msg.Data, &ask.Request) == nil { + r := ask.Request + ask.ID, ask.Repository, ask.Path, ask.Ref = r.ID, r.Repository, r.Path, r.Ref + if at, ok := BuildAskedAt(r.ID); ok { + ask.AskedAt = at + } + } else { + // Nothing read is nothing kept: a half-read request would pass for a delivery (novox/hq issue 558). + ask.Request = BuildRequest{} + } + return ask +} + // WithoutPassword is a repository as a listing may say it: a URL's password, should one ever be written // into it, left out (novox/hq issue 455). Only for what is shown: the ask itself keeps what it was given, // which a cancel's outcome and a rebuild's match are read from. diff --git a/internal/link/queue_order.go b/internal/link/queue_order.go new file mode 100644 index 00000000..7d18712b --- /dev/null +++ b/internal/link/queue_order.go @@ -0,0 +1,243 @@ +package link + +import ( + "context" + "errors" + "fmt" + "sort" + "strings" + "sync" + "time" + + "github.com/nats-io/nats.go" + "github.com/nats-io/nats.go/jetstream" + + "github.com/novox/mesh-controller/internal/broker" +) + +// The order the build queue hands asks out in (novox/hq issue 558). +// +// **A build a walk waits for goes before a pull request's check.** The queue was first come, first served, +// and a merged change's delivery build waited minutes behind checks while its walk, and every walk after +// it, waited on the build. So: +// +// - a delivery is handed out before any check; +// - a check that has waited CheckWaitsAtMost goes before every delivery asked after it, so no check +// waits for ever behind a stream of deliveries; +// - within a kind, first come, first served, by when it was asked; +// - an ask whose kind cannot be told counts as a check: a delivery taken late breaks a budget, a check +// taken late breaks nothing. +// +// **Kept in the stream, never beside it.** The queue is the seat's work queue, which the server hands out +// in stream order to whichever holder pulls; nothing else holds the order, so a restarted controller +// finds it as it was left. An ask that must go behind another is moved to the tail — deleted from where +// it was and published again, its body unchanged, under the same id — while the worker is paused, so no +// holder takes it in the moment it moves. Only waiting asks move: what a holder took is never touched. + +// The kinds of ask, as the queue says them. +const ( + KindDelivery = "delivery" + KindCheck = "check" +) + +// CheckWaitsAtMost is how long a check waits before it goes ahead of the deliveries asked after it. +var CheckWaitsAtMost = 10 * time.Minute + +// KindOf is an ask's kind: a check for a pull request's check — a recheck the merge tool asks included — +// a dry run, or an ask nothing could be read from; a delivery for any other build. +func KindOf(r BuildRequest) string { + if r.ID == "" || r.Check != nil || r.DryRun { + return KindCheck + } + return KindDelivery +} + +// askedOrStored is when an ask was asked, as its id says, or else when the queue took it. +func (a QueuedAsk) askedOrStored() time.Time { + if !a.AskedAt.IsZero() { + return a.AskedAt + } + return a.Stored +} + +// QueueOrder is the order the waiting asks are to be handed out in, now: the deliveries and every check +// that has waited CheckWaitsAtMost, by when each was asked, then the other checks, by when each was asked. +func QueueOrder(waiting []QueuedAsk, now time.Time) []QueuedAsk { + var first, then []QueuedAsk + for _, a := range waiting { + if KindOf(a.Request) == KindCheck && now.Sub(a.askedOrStored()) < CheckWaitsAtMost { + then = append(then, a) + continue + } + first = append(first, a) + } + byAsking := func(s []QueuedAsk) { + sort.SliceStable(s, func(i, j int) bool { + if ti, tj := s[i].askedOrStored(), s[j].askedOrStored(); !ti.Equal(tj) { + return ti.Before(tj) + } + return s[i].Seq < s[j].Seq + }) + } + byAsking(first) + byAsking(then) + return append(first, then...) +} + +// toMove is the asks to move to the tail, in this order, for the queue to stand as order says: everything +// after the longest start of order that the queue already holds in that order. +func toMove(queue, order []QueuedAsk) []QueuedAsk { + at := make(map[uint64]int, len(queue)) + for i, a := range queue { + at[a.Seq] = i + } + kept, last := 0, -1 + for kept < len(order) { + i, ok := at[order[kept].Seq] + if !ok || i < last { + break + } + last = i + kept++ + } + return order[kept:] +} + +// orderPasses keeps one pass at a time in this process: two would each resume the worker the other paused. +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 + +// 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. +func PutInOrder(ctx context.Context, js *broker.JetStream, seat string) ([]string, error) { + orderPasses.Lock() + defer orderPasses.Unlock() + api, err := jetstream.New(js.Conn()) + if err != nil { + 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 { + return nil, err + } + // Paused, then read again: what the worker handed out up to the pause is not waiting any more. + if delivered >= 0 { + if _, err := api.PauseConsumer(ctx, worker.Stream, worker.Name, time.Now().Add(pausedFor)); 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) + } + defer func() { + resuming, cancel := context.WithTimeout(context.WithoutCancel(ctx), 5*time.Second) + 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) + } + }() + if waiting, _, err = readWaiting(ctx, api, worker, seat); err != nil { + return nil, err + } + } + var moved []string + for _, a := range toMove(waiting, QueueOrder(waiting, time.Now())) { + ok, err := moveToTheTail(ctx, js, api, worker, seat, a.Seq) + if err != nil { + return moved, err + } + if ok { + moved = append(moved, a.ID) + } + } + 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) { + 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) + } + delivered := int64(-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) + } + 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) + } + var out []QueuedAsk + seq := uint64(max(delivered, 0)) + 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) + } + out = append(out, askOf(msg)) + seq = msg.Sequence + 1 + } + return out, 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 { + info, err := consumer.Info(ctx) + if err != nil { + return false, err + } + if info.Delivered.Stream >= seq { + return false, nil + } + } + msg, err := stream.GetMsg(ctx, 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. + if !strings.HasPrefix(k, "Nats-") { + again.Header[k] = v + } + } + for try := 0; ; try++ { + publishing, cancel := context.WithTimeout(ctx, 10*time.Second) + _, err := js.Context().PublishMsg(again, nats.Context(publishing)) + cancel() + if err == nil { + return true, nil + } + 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) + } + time.Sleep(time.Second) + } +} diff --git a/internal/link/queue_order_test.go b/internal/link/queue_order_test.go new file mode 100644 index 00000000..3372c7d3 --- /dev/null +++ b/internal/link/queue_order_test.go @@ -0,0 +1,124 @@ +package link + +import ( + "context" + "reflect" + "testing" + "time" +) + +// The order the build queue hands asks out in (novox/hq issue 558): a build a walk waits for before any +// pull request's check, a check that has waited ten minutes before every delivery asked after it, and +// first come, first served within a kind. + +func anAsk(seq uint64, id string, asked time.Time, check bool) QueuedAsk { + r := BuildRequest{ID: id, Repository: "/r"} + if check { + r.Check = &CheckRequest{Owner: "novox", Repo: "r"} + } + return QueuedAsk{Seq: seq, ID: id, AskedAt: asked, Request: r, State: AskWaiting} +} + +func idsOf(asks []QueuedAsk) []string { + out := []string{} + for _, a := range asks { + out = append(out, a.ID) + } + return out +} + +func TestADeliveryIsHandedOutBeforeTheChecksAskedBeforeIt(t *testing.T) { + now := time.Date(2026, 10, 12, 12, 0, 0, 0, time.UTC) + waiting := []QueuedAsk{ + anAsk(1, "check-a", now.Add(-3*time.Minute), true), + anAsk(2, "check-b", now.Add(-2*time.Minute), true), + anAsk(3, "delivery-d", now.Add(-time.Minute), false), + } + want := []string{"delivery-d", "check-a", "check-b"} + if got := idsOf(QueueOrder(waiting, now)); !reflect.DeepEqual(got, want) { + t.Fatalf("handed out as %v, want %v", got, want) + } + // Only the checks ahead of it move behind it; the delivery stays where it is. + if got := idsOf(toMove(waiting, QueueOrder(waiting, now))); !reflect.DeepEqual(got, []string{"check-a", "check-b"}) { + t.Fatalf("moved %v", got) + } +} + +func TestACheckThatWaitedTenMinutesGoesBeforeALaterDelivery(t *testing.T) { + now := time.Date(2026, 10, 12, 12, 0, 0, 0, time.UTC) + // As the queue stands after a delivery put the young checks behind it, ten minutes ago and since. + waiting := []QueuedAsk{ + anAsk(5, "delivery-early", now.Add(-12*time.Minute), false), + anAsk(6, "delivery-later", now.Add(-time.Minute), false), + anAsk(7, "check-old", now.Add(-11*time.Minute), true), + anAsk(8, "check-young", now.Add(-9*time.Minute), true), + } + want := []string{"delivery-early", "check-old", "delivery-later", "check-young"} + if got := idsOf(QueueOrder(waiting, now)); !reflect.DeepEqual(got, want) { + t.Fatalf("handed out as %v, want %v", got, want) + } + if got := idsOf(toMove(waiting, QueueOrder(waiting, now))); !reflect.DeepEqual(got, []string{"delivery-later", "check-young"}) { + t.Fatalf("moved %v", got) + } +} + +func TestAnAskWhoseKindCannotBeToldIsACheck(t *testing.T) { + for _, r := range []BuildRequest{ + {}, // unreadable: nothing was read from it + {ID: "build-1", Check: &CheckRequest{}}, // a pull request's check, or a recheck the merge tool asked + {ID: "build-2", DryRun: true}, // looked at, never delivered + } { + if got := KindOf(r); got != KindCheck { + t.Errorf("%+v is a %s", r, got) + } + } + if got := KindOf(BuildRequest{ID: "build-3", Repository: "/r"}); got != KindDelivery { + t.Errorf("a build is a %s", got) + } +} + +// Against a real server: two checks wait, a delivery is asked, and the next free holder takes the delivery. +func TestNatsADeliveryIsTakenBeforeTheChecksQueuedAheadOfIt(t *testing.T) { + js := aBusWithTheBuildRole(t) + ctx, stop := context.WithCancel(context.Background()) + defer stop() + asks := BuildsOn(js, TheBuildMachine) + now := time.Now() + for i, r := range []BuildRequest{ + {ID: NewBuildID(now.Add(-3 * time.Minute)), Repository: "/a", Check: &CheckRequest{Owner: "novox", Repo: "a"}}, + {ID: NewBuildID(now.Add(-2 * time.Minute)), Repository: "/b", Check: &CheckRequest{Owner: "novox", Repo: "b"}}, + {ID: NewBuildID(now), Repository: "/d"}, + } { + if i == 2 { + // And one nobody can read, whose kind cannot be told: it waits with the checks. + if _, err := js.Context().Publish(BuildWork(), []byte("not an ask")); err != nil { + t.Fatal(err) + } + } + if err := asks.Ask(ctx, r); err != nil { + t.Fatal(err) + } + } + took := make(chan string, 4) + machine := MachineOverNATS(js, "anchor") + defer machine.Close() + go func() { + _ = machine.Take(ctx, func(ctx context.Context, work Build) { + took <- work.Request().Repository + _ = work.Announce(ctx, BuildResult{ID: work.Request().ID, On: "anchor", Failed: "no"}) + _ = work.Done() + }) + }() + var order []string + for len(order) < 3 { + select { + case r := <-took: + order = append(order, r) + case <-time.After(10 * time.Second): + t.Fatalf("taken so far: %v", order) + } + } + if want := []string{"/d", "/a", "/b"}; !reflect.DeepEqual(order, want) { + t.Fatalf("taken as %v, want %v", order, want) + } +} -- 2.54.0 From 1ded7dff34cee1f81af397572ffa6f72e10d688d Mon Sep 17 00:00:00 2001 From: jochen Date: Mon, 12 Oct 2026 21:38:11 +0200 Subject: [PATCH 2/3] 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. --- cmd/mesh-controller/queue.go | 78 ++++---- cmd/mesh-controller/queue_answer_test.go | 13 ++ internal/link/builds_nats.go | 17 +- internal/link/queue.go | 6 +- internal/link/queue_order.go | 233 +++++++++++++++++----- internal/link/queue_order_test.go | 240 +++++++++++++++++++++++ 6 files changed, 489 insertions(+), 98 deletions(-) diff --git a/cmd/mesh-controller/queue.go b/cmd/mesh-controller/queue.go index 9ef17397..5f0c2308 100644 --- a/cmd/mesh-controller/queue.go +++ b/cmd/mesh-controller/queue.go @@ -14,7 +14,6 @@ import ( "time" "github.com/nats-io/nats.go" - "github.com/nats-io/nats.go/jetstream" "github.com/novox/mesh-controller/internal/broker" "github.com/novox/mesh-controller/internal/inventory" @@ -81,37 +80,28 @@ func listQueue(ctx context.Context, args []string, w io.Writer) error { return nil } -// keepingTheQueueInOrder puts the build queue in order every watchEvery while this controller acts -// (novox/hq issue 558). An ask puts it in order as it joins; this is for the clock: a check comes of age -// with nothing asked. +// keepingTheQueueInOrder runs the build queue's passes while this controller acts (novox/hq issue 558): +// the one process that puts the queue in order, so no two passes pause and resume it over each other. +// The build seat is read from the catalogue at most every half minute. func keepingTheQueueInOrder(ctx context.Context, open *stores, js *broker.JetStream) { - tick := time.NewTicker(watchEvery) - defer tick.Stop() - for { - select { - case <-ctx.Done(): - return - case <-tick.C: - } - if !link.Holding() || js == nil { - continue - } - entries, err := open.inventory.Catalogued(ctx) - if err != nil { - fmt.Printf("the build queue is not put in order: the catalogue cannot be read: %v\n", err) - continue - } - seat := buildSeatAmong(entries) - ordering, cancel := context.WithTimeout(ctx, watchEvery) - moved, err := link.PutInOrder(ordering, js, seat) - cancel() - if len(moved) > 0 { - fmt.Printf("the %s queue put in order: %s moved behind what goes first\n", seat, strings.Join(moved, ", ")) - } - if err != nil { - fmt.Printf("the %s queue could not be put in order: %v\n", seat, err) - } + if js == nil { + return } + seat, readAt := link.TheBuildMachine, time.Time{} + seatNow := func() string { + if time.Since(readAt) < watchEvery { + return seat + } + if entries, err := open.inventory.Catalogued(ctx); err == nil { + seat, readAt = buildSeatAmong(entries), time.Now() + } else { + fmt.Printf("the build seat is read as %s: the catalogue cannot be read: %v\n", seat, err) + } + return seat + } + link.KeepingInOrder(ctx, js, seatNow, link.Holding, func(format string, args ...any) { + fmt.Printf(format+"\n", args...) + }) } // queueAnswer is the queue as data (novox/hq issue 455): what a caller reads to say how far back an ask @@ -307,6 +297,15 @@ func cancelAsk(ctx context.Context, js *broker.JetStream, open *stores, seat str } } worker, _ := broker.HolderConsumerFor("", "", broker.DeclaredSeat{Name: seat, Accepts: []string{"build"}}) + // **Where it waits now** (novox/hq issue 558): moved into order since it was read, a waiting ask has a + // later sequence, and the worker handing out what was behind its old one is not it being taken. + if ask.State == link.AskWaiting { + if q, err := link.ReadQueue(ctx, js, seat); err == nil { + if now, ok := q.Find(ask.ID); ok && now.State == link.AskWaiting { + ask.Seq = now.Seq + } + } + } switch ask.State { case link.AskWaiting: if taken, err := takenSince(js, worker, ask.Seq); err != nil { @@ -328,14 +327,25 @@ func cancelAsk(ctx context.Context, js *broker.JetStream, open *stores, seat str "ends it there", ask.ID, started, ask.ID) } } - if err := js.Context().DeleteMsg(worker.Stream, ask.Seq); err != nil && !errors.Is(err, nats.ErrMsgNotFound) && - !errors.Is(err, jetstream.ErrMsgNotFound) && !strings.Contains(err.Error(), "no message found") { + // Found by its id when it was moved into order since it was read (novox/hq issue 558). + removed, err := link.DeleteAsk(ctx, js, seat, ask) + if err != nil { return "", fmt.Errorf("%s is marked cancelled and could not be deleted from the queue (seq %d): %w — a "+ "holder taking it ends it as cancelled", ask.ID, ask.Seq, err) } recordCancelled(ctx, open, ask) - return fmt.Sprintf("cancelled %s (%s, %s): deleted from the %s queue and recorded failed, %s — a plan "+ - "that asked for it fails with that", ask.ID, ask.Repository, ask.State, seat, link.CancelledByHand), nil + return cancelSaid(ask, seat, removed), nil +} + +// cancelSaid is what a cancel says: deleted from the queue only when it was. +func cancelSaid(ask link.QueuedAsk, seat string, removed uint64) string { + where := fmt.Sprintf("deleted from the %s queue", seat) + if removed == 0 { + where = fmt.Sprintf("no longer in the %s queue to delete, and marked cancelled, so a holder taking it "+ + "ends it", seat) + } + return fmt.Sprintf("cancelled %s (%s, %s): %s and recorded failed, %s — a plan "+ + "that asked for it fails with that", ask.ID, ask.Repository, ask.State, where, link.CancelledByHand) } // holderLooks is how long a cancel waits for a holder that took the ask before the mark to say it diff --git a/cmd/mesh-controller/queue_answer_test.go b/cmd/mesh-controller/queue_answer_test.go index ca6fed93..6f8cf6c9 100644 --- a/cmd/mesh-controller/queue_answer_test.go +++ b/cmd/mesh-controller/queue_answer_test.go @@ -142,3 +142,16 @@ func TestTheQueueNeverShowsARepositorysPassword(t *testing.T) { } } } + +// A cancel says it deleted the ask only when it did: moved into order and gone since, it says so (novox/hq +// issue 558). +func TestACancelSaysDeletedOnlyWhenItDeleted(t *testing.T) { + ask := link.QueuedAsk{ID: "build-1", Repository: "novox/x", State: link.AskWaiting, Seq: 4} + if said := cancelSaid(ask, link.TheBuildMachine, 9); !strings.Contains(said, "deleted from the") { + t.Errorf("removed, it says: %s", said) + } + if said := cancelSaid(ask, link.TheBuildMachine, 0); strings.Contains(said, "deleted from") || + !strings.Contains(said, "no longer in the") { + t.Errorf("gone already, it says: %s", said) + } +} diff --git a/internal/link/builds_nats.go b/internal/link/builds_nats.go index 3e11fca2..9201fe93 100644 --- a/internal/link/builds_nats.go +++ b/internal/link/builds_nats.go @@ -79,21 +79,11 @@ func (b *natsBuilds) Ask(ctx context.Context, request BuildRequest) error { if _, err := b.js.Context().Publish(BuildWorkOf(b.role()), body, nats.Context(publish)); err != nil { return fmt.Errorf("cannot submit a build: %w", err) } - b.putInOrder(ctx) + // The acting controller puts it in order (novox/hq issue 558); this only says an ask joined. + signalAsked() return nil } -// putInOrder puts the queue in order after an ask joined it (novox/hq issue 558): a delivery goes ahead of -// the checks waiting before it now, not when the serving controller next looks. Never the ask's failure: -// the ask is in the queue, and a queue left first come, first served still answers it. -func (b *natsBuilds) putInOrder(ctx context.Context) { - ordering, cancel := context.WithTimeout(ctx, 15*time.Second) - defer cancel() - if _, err := PutInOrder(ordering, b.js, b.role()); err != nil { - fmt.Fprintf(os.Stderr, "the %s queue could not be put in order: %v\n", b.role(), err) - } -} - // ErrNotHandedOver is a Submit that failed before the build was handed to the build seat: nothing runs. Any // other error from Submit came after: the build may still run, and its outcome may still come (novox/hq // issue 325). @@ -125,7 +115,8 @@ func (b *natsBuilds) Submit(ctx context.Context, request BuildRequest, if _, err := b.js.Context().Publish(BuildWorkOf(b.role()), body, nats.Context(publish)); err != nil { return BuildResult{}, fmt.Errorf("%w: cannot submit a build: %w", ErrNotHandedOver, err) } - b.putInOrder(ctx) + // The acting controller puts it in order (novox/hq issue 558); this only says an ask joined. + signalAsked() waiting, cancelWait := context.WithTimeout(ctx, wait) defer cancelWait() diff --git a/internal/link/queue.go b/internal/link/queue.go index b70cc7d3..a7c87652 100644 --- a/internal/link/queue.go +++ b/internal/link/queue.go @@ -62,7 +62,7 @@ type QueuedAsk struct { Path string `json:"path,omitempty"` Ref string `json:"ref,omitempty"` AskedAt time.Time `json:"asked-at,omitempty"` - // Stored is when the stream took it: when it was asked, or last moved into order (novox/hq issue 558). + // 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"` // On and Started are the holder building an in-flight ask and when it started, from its started @@ -266,6 +266,10 @@ func ReadQueue(ctx context.Context, js *broker.JetStream, seat string) (Queue, e // askOf is one ask as the stream holds it: what its request says, and when the stream took it. func askOf(msg *jetstream.RawStreamMsg) QueuedAsk { ask := QueuedAsk{Seq: msg.Sequence, Stored: msg.Time} + // Moved into order, it keeps when the stream first took it (novox/hq issue 558). + if at, err := time.Parse(time.RFC3339Nano, msg.Header.Get(QueuedAtHeader)); err == nil { + ask.Stored = at + } if json.Unmarshal(msg.Data, &ask.Request) == nil { r := ask.Request ask.ID, ask.Repository, ask.Path, ask.Ref = r.ID, r.Repository, r.Path, r.Ref diff --git a/internal/link/queue_order.go b/internal/link/queue_order.go index 7d18712b..c776dc42 100644 --- a/internal/link/queue_order.go +++ b/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 +} diff --git a/internal/link/queue_order_test.go b/internal/link/queue_order_test.go index 3372c7d3..efcf9284 100644 --- a/internal/link/queue_order_test.go +++ b/internal/link/queue_order_test.go @@ -2,9 +2,16 @@ package link import ( "context" + "encoding/json" + "fmt" "reflect" + "sync" "testing" "time" + + "github.com/nats-io/nats.go" + + "github.com/novox/mesh-controller/internal/broker" ) // The order the build queue hands asks out in (novox/hq issue 558): a build a walk waits for before any @@ -99,6 +106,10 @@ func TestNatsADeliveryIsTakenBeforeTheChecksQueuedAheadOfIt(t *testing.T) { t.Fatal(err) } } + // The acting controller's loop puts it in order, told by the asks. + go KeepingInOrder(ctx, js, func() string { return TheBuildMachine }, func() bool { return true }, + func(string, ...any) {}) + time.Sleep(500 * time.Millisecond) took := make(chan string, 4) machine := MachineOverNATS(js, "anchor") defer machine.Close() @@ -122,3 +133,232 @@ func TestNatsADeliveryIsTakenBeforeTheChecksQueuedAheadOfIt(t *testing.T) { t.Fatalf("taken as %v, want %v", order, want) } } + +// --- the moves, against a real server ------------------------------------------------------------ + +func publishAsk(t *testing.T, js *broker.JetStream, r BuildRequest) { + t.Helper() + body, _ := json.Marshal(r) + if _, err := js.Context().Publish(BuildWork(), body); err != nil { + t.Fatal(err) + } +} + +func aCheck(asked time.Time, repository string) BuildRequest { + return BuildRequest{ID: NewBuildID(asked), Repository: repository, Check: &CheckRequest{Owner: "novox", Repo: "r"}} +} + +func aDelivery(asked time.Time, repository string) BuildRequest { + return BuildRequest{ID: NewBuildID(asked), Repository: repository} +} + +// queued is the stream's asks, in queue order. +func queued(t *testing.T, js *broker.JetStream) []QueuedAsk { + t.Helper() + q, err := ReadQueue(context.Background(), js, TheBuildMachine) + if err != nil { + t.Fatal(err) + } + return q.Asks +} + +// takeAll runs a holder that records how often it was handed each repository. +func takeAll(t *testing.T, ctx context.Context, js *broker.JetStream) (chan string, func() map[string]int) { + t.Helper() + took := make(chan string, 100) + var mu sync.Mutex + counts := map[string]int{} + machine := MachineOverNATS(js, "anchor") + t.Cleanup(machine.Close) + go func() { + _ = machine.Take(ctx, func(ctx context.Context, work Build) { + mu.Lock() + counts[work.Request().Repository]++ + mu.Unlock() + took <- work.Request().Repository + _ = work.Announce(ctx, BuildResult{ID: work.Request().ID, On: "anchor", Failed: "no"}) + _ = work.Done() + }) + }() + return took, func() map[string]int { + mu.Lock() + defer mu.Unlock() + out := map[string]int{} + for k, v := range counts { + out[k] = v + } + return out + } +} + +// A move that published its copy and stopped before deleting the original leaves two copies of one ask; +// the next pass makes them one, and a copy of one a holder already took is removed, never built twice. +func TestNatsAMoveCutShortLeavesOneCopyAfterTheNextPass(t *testing.T) { + js := aBusWithTheBuildRole(t) + ctx := t.Context() + now := time.Now() + 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) + } + var got []string + for _, a := range queued(t, js) { + got = append(got, a.Repository) + } + if want := []string{"/d", "/c"}; !reflect.DeepEqual(got, want) { + t.Fatalf("after the pass the queue holds %v, want %v", got, want) + } + + // And one a holder took before the copy was published: the copy goes, the holder's stays. + taken := aCheck(now, "/taken") + publishAsk(t, js, taken) + 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() }() + // Take everything waiting up to and including /taken, and keep /taken unsettled. + for { + 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.Repository == "/taken" { + break + } + _ = msgs[0].Ack() + } + publishAsk(t, js, taken) + if _, err := PutInOrder(ctx, js, TheBuildMachine); err != nil { + t.Fatal(err) + } + n := 0 + for _, a := range queued(t, js) { + if a.ID == taken.ID { + n++ + } + } + if n != 1 { + t.Fatalf("%d copies of the ask a holder took are in the queue, want 1", n) + } +} + +// Two passes at once: nothing is built twice, and nothing is lost. +func TestNatsTwoPassesAtOnceBuildNothingTwice(t *testing.T) { + js := aBusWithTheBuildRole(t) + ctx, stop := context.WithCancel(t.Context()) + defer stop() + took, counts := takeAll(t, ctx, js) + now := time.Now() + const asks = 24 + var passes sync.WaitGroup + for p := 0; p < 2; p++ { + passes.Add(1) + go func() { + defer passes.Done() + for i := 0; i < 15; i++ { + _, _ = PutInOrder(ctx, js, TheBuildMachine) + } + }() + } + for i := 0; i < asks; i++ { + repository := fmt.Sprintf("/%02d", i) + if i%3 == 2 { + publishAsk(t, js, aDelivery(now.Add(time.Duration(i)*time.Millisecond), repository)) + } else { + publishAsk(t, js, aCheck(now.Add(time.Duration(i)*time.Millisecond), repository)) + } + } + passes.Wait() + for got := 0; got < asks; got++ { + select { + case <-took: + case <-time.After(15 * time.Second): + t.Fatalf("taken %d of %d: %v", got, asks, counts()) + } + } + time.Sleep(time.Second) + c := counts() + for i := 0; i < asks; i++ { + if n := c[fmt.Sprintf("/%02d", i)]; n != 1 { + t.Errorf("/%02d was built %d times", i, n) + } + } +} + +// A check that has waited ten minutes goes before a delivery asked after it, even one queued ahead of it. +func TestNatsACheckThatWaitedTenMinutesIsTakenBeforeALaterDelivery(t *testing.T) { + js := aBusWithTheBuildRole(t) + ctx, stop := context.WithCancel(t.Context()) + defer stop() + now := time.Now() + publishAsk(t, js, aDelivery(now.Add(-time.Minute), "/delivery")) + publishAsk(t, js, aCheck(now.Add(-11*time.Minute), "/old-check")) + if _, err := PutInOrder(ctx, js, TheBuildMachine); err != nil { + t.Fatal(err) + } + took, _ := takeAll(t, ctx, js) + var order []string + for len(order) < 2 { + select { + case r := <-took: + order = append(order, r) + case <-time.After(10 * time.Second): + t.Fatalf("taken so far: %v", order) + } + } + if want := []string{"/old-check", "/delivery"}; !reflect.DeepEqual(order, want) { + t.Fatalf("taken as %v, want %v", order, want) + } +} + +// A cancel that read an ask before it moved deletes the moved copy, found by its id, and says when +// there was nothing left to delete. +func TestNatsACancelRacingAMoveDeletesTheMovedCopy(t *testing.T) { + js := aBusWithTheBuildRole(t) + ctx := t.Context() + now := time.Now() + 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) + } + removed, err := DeleteAsk(ctx, js, TheBuildMachine, read) + if err != nil || removed == 0 || removed == read.Seq { + t.Fatalf("removed seq %d (read at %d): %v", removed, read.Seq, err) + } + for _, a := range queued(t, js) { + if a.ID == read.ID { + t.Fatalf("the cancelled ask is still queued at seq %d", a.Seq) + } + } + if removed, err := DeleteAsk(ctx, js, TheBuildMachine, read); err != nil || removed != 0 { + t.Fatalf("an ask gone already is said removed at %d: %v", removed, err) + } +} + +// An ask nothing can be read from keeps its place in time when it moves. +func TestNatsAnUnreadableAskKeepsItsAgeWhenItMoves(t *testing.T) { + js := aBusWithTheBuildRole(t) + ctx := t.Context() + if _, err := js.Context().Publish(BuildWork(), []byte("not an ask")); err != nil { + t.Fatal(err) + } + 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) + } + asks := queued(t, js) + if len(asks) != 2 || asks[1].ID != "" || !asks[1].Stored.Equal(first) { + t.Fatalf("after the move the unreadable ask was taken at %v, first at %v: %+v", asks[1].Stored, first, asks) + } +} -- 2.54.0 From fc573f951a4731ac3b8d1c4fcd03b29058cef787 Mon Sep 17 00:00:00 2001 From: jochen Date: Mon, 12 Oct 2026 21:46:49 +0200 Subject: [PATCH 3/3] 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) { -- 2.54.0