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) + } +}