Hand a delivery build out before pull-request checks (issue 558) #276

Merged
mesh-admin merged 3 commits from fix/558-a-delivery-build-goes-first into main 2026-10-12 21:20:19 +00:00
7 changed files with 978 additions and 26 deletions
+56 -14
View File
@@ -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,6 +80,30 @@ func listQueue(ctx context.Context, args []string, w io.Writer) error {
return nil
}
// 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) {
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
// 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 +123,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 +153,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 +210,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 {
@@ -275,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 {
@@ -296,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
+22 -2
View File
@@ -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 {
@@ -135,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)
}
}
+3
View File
@@ -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)
+4
View File
@@ -79,6 +79,8 @@ 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)
}
// The acting controller puts it in order (novox/hq issue 558); this only says an ask joined.
signalAsked()
return nil
}
@@ -113,6 +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)
}
// 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()
+26 -10
View File
@@ -62,7 +62,11 @@ 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 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.
@@ -236,15 +240,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 +265,26 @@ 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}
// 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
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.
+426
View File
@@ -0,0 +1,426 @@
package link
import (
"context"
"errors"
"fmt"
"sort"
"strconv"
"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. 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. 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
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)
// 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. 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) (Pass, error) {
var pass Pass
orderPasses.Lock()
defer orderPasses.Unlock()
api, err := jetstream.New(js.Conn())
if err != nil {
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 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 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() {
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 by %s: %v\n", seat, pauseEnds.Format(time.RFC3339), err)
}
}()
if waiting, extra, _, err = readWaiting(ctx, api, worker, seat); err != nil {
return pass, err
}
}
stream, err := api.Stream(ctx, worker.Stream)
if err != nil {
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 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))
}
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.
break
}
ok, err := moveToTheTail(ctx, js, stream, worker.Name, seat, a)
if err != nil {
return pass, err
}
if ok {
pass.Moved = append(pass.Moved, a.ID)
}
}
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 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)
if err != nil {
return nil, nil, 0, fmt.Errorf("the %s work queue (%s) cannot be read: %w", seat, worker.Stream, err)
}
delivered = -1
if consumer, err := stream.Consumer(ctx, worker.Name); err == nil {
info, err := consumer.Info(ctx)
if err != nil {
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, 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.
// 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)))
if errors.Is(err, jetstream.ErrMsgNotFound) {
break
}
if err != nil {
return nil, nil, 0, fmt.Errorf("reading the %s work queue: %w", seat, err)
}
seq = msg.Sequence + 1
a := askOf(msg)
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:
extra = append(extra, a)
default:
waiting = append(waiting, a)
}
}
return waiting, extra, delivered, 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 >= a.Seq {
return false, nil
}
}
msg, err := stream.GetMsg(ctx, a.Seq)
if errors.Is(err, jetstream.ErrMsgNotFound) {
return false, nil
}
if err != 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
}
}
// 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))
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)
}
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)
pass, err := PutInOrder(passing, js, name)
cancel()
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)
}
if info, err := js.Context().StreamInfo(worker.Stream); err == nil {
// Its own moves are not an ask.
lastSeq = info.State.LastSeq
}
}
}
// 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
}
+441
View File
@@ -0,0 +1,441 @@
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
// 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)
}
}
// 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()
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)
}
}
// --- 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"))
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) {
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()
}
moveCopyOf(t, js, taken.ID)
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)
}
}
// 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)
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 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 {
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 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) {
t.Fatalf("after the move the unreadable ask was taken at %v, first at %v: %+v", asks[1].Stored, first, asks)
}
}