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