novox/mesh-controller
13
0
Fork 0
Code Issues Pull Requests 4 Releases Wiki Activity

issue 974: the checks are queued by pull request on two work-queue streams, taken after a delivery build, dropped when judged, stopped when their key moves; the reordering and the forge-asking end are gone

Browse Source
This commit is contained in:
jschoubben
2026-10-17 14:01:33 +02:00
parent f883830975
commit 3d09d5235f
20 changed files with 888 additions and 1412 deletions
Download Patch File Download Diff File Expand all files Collapse all files
internal/link/build.go
+6
View File
@@ -144,6 +144,12 @@ type CheckRequest struct {
Repo string `json:"repo"`
Number int `json:"number,omitempty"`
Base string `json:"base,omitempty"`
// Trunk is the base branch's head the check is asked against, as the forge announced it with the head (novox/hq ADR
// 0406 §4): with the head, what the pull request's key says the current check is for, and what a waiting request
// is dropped by and a running check stopped by when the key moves.
Trunk string `json:"trunk,omitempty"`
// Ready says the pull request is marked ready (ADR 0406 §3): its check is queued before every other check.
Ready bool `json:"ready,omitempty"`
// Paths are the files the pull request changes, for the width of its rebuild.
Paths []string `json:"paths,omitempty"`
// Beside are the repositories the check reads next to this one, each cloned at the ref given — the
internal/link/builds_nats.go
+187 -74
View File
@@ -76,11 +76,9 @@ func (b *natsBuilds) Ask(ctx context.Context, request BuildRequest) error {
}
publish, cancel := context.WithTimeout(ctx, 30*time.Second)
defer cancel()
if _, err := b.js.Context().Publish(BuildWorkOf(b.role()), body, nats.Context(publish)); err != nil {
if _, err := b.js.Context().Publish(WorkSubjectOf(b.role(), request), 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
}
@@ -112,11 +110,9 @@ func (b *natsBuilds) Submit(ctx context.Context, request BuildRequest,
// be assumed, because nothing else will ever say so.
publish, cancel := context.WithTimeout(ctx, 30*time.Second)
defer cancel()
if _, err := b.js.Context().Publish(BuildWorkOf(b.role()), body, nats.Context(publish)); err != nil {
if _, err := b.js.Context().Publish(WorkSubjectOf(b.role(), request), 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()
@@ -162,10 +158,29 @@ type natsMachine struct {
js *broker.JetStream
on string
seat string
sub *nats.Subscription
// subs are the queues this machine takes from, in the order it takes from them: the seat's own, where a delivery
// build waits, then the checks of pull requests marked ready, then every other check (novox/hq ADR 0406 §3).
subs []*takeQueue
opts MachineOptions
}
// takeQueue is one queue a machine pulls from.
type takeQueue struct {
sub *nats.Subscription
// kind is what its asks are: a build, a ready check, or another check.
kind string
// worker is the consumer bound; bound is false until a bind succeeded, for a queue the mesh may not have made yet.
worker broker.Consumer
bound bool
}
// The kinds of queue a machine takes from, in the order it takes from them.
const (
queueBuilds = "builds"
queueReady = "ready"
queueOther = "other"
)
// MachineOptions is what a holder tells the taking loop about itself (novox/hq ADR 0219).
type MachineOptions struct {
// Paused is asked before every fetch: while it says so, nothing new is taken, and a build
@@ -207,47 +222,56 @@ const (
)
func (m *natsMachine) Close() {
if m.sub != nil {
_ = m.sub.Unsubscribe()
for _, q := range m.subs {
if q.sub != nil {
_ = q.sub.Unsubscribe()
}
}
}
// Take binds to the role's worker and pulls one request at a time, handing each over.
// Take binds to the role's workers and pulls one request at a time, handing each over.
//
// **Bound, never created.** The work queue and the worker on it are the controller's to define
// **Bound, never created.** The work queues and the workers on them are the controller's to define
// (design 25 §3), and a build machine reaches no part of the JetStream API — so a missing one is said
// as the mesh's to answer rather than quietly created with whatever this client defaults to.
//
// **Pulled, one at a time, by whichever holder is free** (novox/hq ADR 0190). Every machine holding
// the role binds this same worker; a machine asks for the next request only when it has finished
// the role binds these same workers; a machine asks for the next request only when it has finished
// the last, so a slow machine never holds an ask an idle one could take, and a machine that took
// five at once would run five container builds against one runtime and finish all of them slower
// than the first.
//
// **In this order** (novox/hq ADR 0406 §3): a delivery build, then a check of a pull request marked ready, then any
// other check, each queue tried without waiting; when all three are empty the first is waited on for a second at most
// and the three are tried again, so a delivery build is never much more than a second behind and nothing spins.
func (m *natsMachine) Take(ctx context.Context, do func(context.Context, Build)) error {
worker, found := broker.HolderConsumerFor(m.on, "build-agent",
broker.DeclaredSeat{Name: m.seat, Accepts: []string{"build"}})
if !found {
return fmt.Errorf("%s accepts no work, so there is nothing for this machine to take", m.seat)
}
// **The consumer's own filter, not the one subject this machine cares about.** The client checks
// what is asked for against the consumer's filter and refuses anything that is not the same —
// "subject does not match consumer" — so subscribing `…accept.build` against a consumer filtered
// on `…accept.>` is rejected even though it is narrower. Learned twice now, on two different
// consumers, which is why it is written down here.
filter := worker.Filters[0]
sub, err := m.js.Context().PullSubscribe(filter, worker.Name,
nats.Bind(worker.Stream, worker.Name), nats.ManualAck())
if err != nil {
m.subs = []*takeQueue{{kind: queueBuilds, worker: worker}}
if m.seat == broker.TheBuildSeat {
for i, w := range broker.CheckWorkers() {
kind := queueReady
if i == 1 {
kind = queueOther
}
m.subs = append(m.subs, &takeQueue{kind: kind, worker: w})
}
}
// The seat's own queue is bound at once and its absence is this machine's error, as it always was. The check queues
// are bound when they are there: a controller that has not made them yet is no reason to take no build.
if err := m.bind(m.subs[0]); err != nil {
return fmt.Errorf(
"this machine cannot take work from %s: %w. The mesh creates that queue and this "+
"machine's worker on it, and a build machine may not create one itself — so this is "+
"the mesh's to answer, not this machine's", m.seat, err)
}
m.sub = sub
if m.opts.Taking != nil {
m.opts.Taking()
}
var bindAgain time.Time
for {
if ctx.Err() != nil {
@@ -264,29 +288,31 @@ func (m *natsMachine) Take(ctx context.Context, do func(context.Context, Build))
}
continue
}
// One, and wait a while for it; an empty queue is a timeout, which is the normal state of a
// machine with nothing to build, and is asked again.
// Asked for a few seconds at a time when the holder can be paused, so a pause reaches a pull
// already waiting within that, rather than when the client's own wait runs out.
asking, endAsking := ctx, func() {}
if m.opts.Paused != nil {
asking, endAsking = context.WithTimeout(ctx, pausedFetch)
if time.Now().After(bindAgain) {
for _, q := range m.subs[1:] {
if !q.bound {
if err := m.bind(q); err != nil {
fmt.Fprintf(os.Stderr, "the %s check queue is not there to take from yet: %v\n", q.kind, err)
}
}
}
bindAgain = time.Now().Add(30 * time.Second)
}
fetched, err := sub.Fetch(1, nats.Context(asking))
endAsking()
msg, q, err := m.next(ctx)
switch {
case ctx.Err() != nil:
// Ours ended: the machine is being stopped.
return nil
case errors.Is(err, context.Canceled), errors.Is(err, context.DeadlineExceeded), errors.Is(err, nats.ErrTimeout):
case msg != nil:
case errors.Is(err, context.Canceled), errors.Is(err, context.DeadlineExceeded), errors.Is(err, nats.ErrTimeout), err == nil:
// **An empty queue, not the end.** A fetch on a context without a deadline waits the
// client's own while and then says the deadline passed — the client's, not ours. Read
// as "stop", every idle build machine exited clean every half minute and was started
// again by its supervisor, which looked like a crash loop with nothing in the log to
// say why (2026-10-03, the first build agents). Asked again.
continue
case err != nil:
if sub.IsValid() {
default:
if m.subs[0].sub.IsValid() {
// A transient fault in asking — a reconnect, a slow server — is asked past rather
// than ending the machine; one that outlasts the ack wait redelivers nothing lost.
time.Sleep(time.Second)
@@ -294,49 +320,130 @@ func (m *natsMachine) Take(ctx context.Context, do func(context.Context, Build))
}
return fmt.Errorf("the bus stopped delivering build work: %w", err)
}
for _, msg := range fetched {
// **Paused while the pull was answered** (ADR 0219): "takes no new build" holds even for
// the one that arrived in that moment. Handed back at once for another holder — counted as
// a delivery, which a pause landing exactly then costs and nothing else does.
if m.opts.Paused != nil && m.opts.Paused() {
_ = msg.Nak()
// **Paused while the pull was answered** (ADR 0219): "takes no new build" holds even for
// the one that arrived in that moment. Handed back at once for another holder — counted as
// a delivery, which a pause landing exactly then costs and nothing else does.
if m.opts.Paused != nil && m.opts.Paused() {
_ = msg.Nak()
continue
}
var request BuildRequest
if err := json.Unmarshal(msg.Data, &request); err != nil {
// Unreadable: terminated rather than retried, because the next attempt reads the same
// bytes. Nobody waiting hears an answer, which is right — there was no request.
_ = msg.Term()
continue
}
// **Cancelled in the moment it was fetched** (novox/hq ADR 0219): the controller wrote the
// id into the seat's cancelled set before deleting the ask, so one taken in between is
// ended here, terminated rather than redelivered, and its outcome said as failed — never
// built. A set that cannot be read is said and the ask built: a cancel is a person's
// exception, and a holder that refused every build while its grant was missing would
// stop the mesh building for it.
build := &natsBuild{request: request, msg: msg, on: m.on, js: m.js, seat: m.seat, delivery: deliveryOf(msg)}
if cancelled, err := IsCancelled(m.js.Conn(), m.seat, request.ID); err != nil {
fmt.Fprintf(os.Stderr, "could not read whether %s was cancelled, so it is built: %v\n", request.ID, err)
} else if cancelled {
endCancelled(ctx, build)
continue
}
// **A check left behind is dropped when it is taken** (novox/hq ADR 0406 §3): a request on the third queue for a
// head and a trunk commit its pull request's key already says were judged is acknowledged and not run.
if q.kind == queueOther {
if why := m.judgedAlready(ctx, request); why != "" {
fmt.Fprintf(os.Stderr, "%s was left behind in the check queue and is dropped: %s\n", request.ID, why)
_ = msg.Ack()
continue
}
var request BuildRequest
if err := json.Unmarshal(msg.Data, &request); err != nil {
// Unreadable: terminated rather than retried, because the next attempt reads the same
// bytes. Nobody waiting hears an answer, which is right — there was no request.
_ = msg.Term()
continue
}
// **Cancelled in the moment it was fetched** (novox/hq ADR 0219): the controller wrote the
// id into the seat's cancelled set before deleting the ask, so one taken in between is
// ended here, terminated rather than redelivered, and its outcome said as failed — never
// built. A set that cannot be read is said and the ask built: a cancel is a person's
// exception, and a holder that refused every build while its grant was missing would
// stop the mesh building for it.
build := &natsBuild{request: request, msg: msg, on: m.on, js: m.js, seat: m.seat, delivery: deliveryOf(msg)}
if cancelled, err := IsCancelled(m.js.Conn(), m.seat, request.ID); err != nil {
fmt.Fprintf(os.Stderr, "could not read whether %s was cancelled, so it is built: %v\n", request.ID, err)
} else if cancelled {
endCancelled(ctx, build)
continue
}
// **Drained in the moment it was fetched** (novox/hq ADR 0335): handed back at once, as a pause
// landing then is.
if m.opts.Claim != nil && !m.opts.Claim() {
_ = msg.Nak()
continue
}
// A build outlives the acknowledgement window many times over; said while it runs,
// as the controller says it for its own long handlers, so the server neither hands
// the ask to a second machine nor counts the wait against its deliveries.
working := make(chan struct{})
go stillWorking(msg, working, heartbeatWhileWorking)
do(ctx, build)
close(working)
}
// **Drained in the moment it was fetched** (novox/hq ADR 0335): handed back at once, as a pause
// landing then is.
if m.opts.Claim != nil && !m.opts.Claim() {
_ = msg.Nak()
continue
}
// A build outlives the acknowledgement window many times over; said while it runs,
// as the controller says it for its own long handlers, so the server neither hands
// the ask to a second machine nor counts the wait against its deliveries.
working := make(chan struct{})
go stillWorking(msg, working, heartbeatWhileWorking)
do(ctx, build)
close(working)
}
}
// bind binds a queue's worker, which exists and is not created here.
//
// **The consumer's own filter, not the one subject this machine cares about.** The client checks what is asked for
// against the consumer's filter and refuses anything that is not the same — "subject does not match consumer" — so
// subscribing `…accept.build` against a consumer filtered on `…accept.>` is rejected even though it is narrower. Learned
// twice now, on two different consumers, which is why it is written down here.
func (m *natsMachine) bind(q *takeQueue) error {
sub, err := m.js.Context().PullSubscribe(q.worker.Filters[0], q.worker.Name,
nats.Bind(q.worker.Stream, q.worker.Name), nats.ManualAck())
if err != nil {
return err
}
q.sub, q.bound = sub, true
return nil
}
// next takes the next request in the order of the queues: each tried for a moment, and when all are empty the first waited
// on for a second. Nothing, with no error, when there was nothing.
func (m *natsMachine) next(ctx context.Context) (*nats.Msg, *takeQueue, error) {
var last error
for _, q := range m.subs {
if !q.bound {
continue
}
trying, end := context.WithTimeout(ctx, checkTakeWait)
fetched, err := q.sub.Fetch(1, nats.Context(trying))
end()
if len(fetched) > 0 {
return fetched[0], q, nil
}
if err != nil && !errors.Is(err, context.DeadlineExceeded) && !errors.Is(err, nats.ErrTimeout) &&
!errors.Is(err, context.Canceled) {
last = err
}
if ctx.Err() != nil {
return nil, nil, nil
}
}
if last != nil {
return nil, nil, last
}
waiting, end := context.WithTimeout(ctx, takeIdleWait)
defer end()
fetched, err := m.subs[0].sub.Fetch(1, nats.Context(waiting))
if len(fetched) > 0 {
return fetched[0], m.subs[0], nil
}
if err != nil && !errors.Is(err, context.DeadlineExceeded) && !errors.Is(err, nats.ErrTimeout) &&
!errors.Is(err, context.Canceled) {
return nil, nil, err
}
return nil, nil, nil
}
// judgedAlready says why a check just taken from the other checks' queue need not run, or nothing: its pull request's key
// holds its head and trunk commit and a verdict. A key that cannot be read, or none, runs the check (the key is
// mesh-delivery's, and a check wrongly dropped is worse than one run twice).
func (m *natsMachine) judgedAlready(ctx context.Context, r BuildRequest) string {
c := r.Check
if c == nil || c.Group != "" || c.Try || c.Number <= 0 {
return ""
}
key, found, err := ReadPullKey(ctx, m.js.Conn(), c.Owner, c.Repo, c.Number)
if err != nil {
fmt.Fprintf(os.Stderr, "could not read the key of %s/%s#%d, so its check runs: %v\n", c.Owner, c.Repo, c.Number, err)
return ""
}
if found && key.Judged(r.Ref, c.Trunk) {
return fmt.Sprintf("%s/%s#%d is judged already at %.12s on %.12s (%s)", c.Owner, c.Repo, c.Number, r.Ref, c.Trunk,
key.Verdict)
}
return ""
}
type natsBuild struct {
@@ -398,6 +505,12 @@ func (b *natsBuild) Announce(ctx context.Context, result BuildResult) error {
func (b *natsBuild) Done() error { return b.msg.Ack() }
// WatchPull has moved called once the pull request's key moves on from what this check was asked for; the stop ends the
// watch (see WatchPull). A request that is no pull request's check watches nothing.
func (b *natsBuild) WatchPull(ctx context.Context, moved func(PullKey)) (func(), error) {
return WatchPull(ctx, b.js.Conn(), b.request, moved)
}
// Began publishes that this machine has taken the build, into the stream like the outcome, so a
// reader that asks afterwards sees when it started as well as how it ended.
func (b *natsBuild) Began(ctx context.Context) error {
internal/link/builds_nats_test.go
+6
View File
@@ -44,6 +44,12 @@ func aBusWithTheBuildRole(t *testing.T) *broker.JetStream {
}); err != nil {
t.Fatal(err)
}
// And the two check queues' workers, which the controller makes beside the streams (novox/hq ADR 0406 §3).
for _, w := range broker.CheckWorkers() {
if err := js.EnsureConsumer(w); err != nil {
t.Fatal(err)
}
}
clean := func() {
_ = js.Context().DeleteStream("SEAT_NODE_BUILD_AGENT")
for _, s := range broker.MeshStreams() {
internal/link/checks_nats.go
+170
View File
@@ -0,0 +1,170 @@
package link
import (
"context"
"encoding/json"
"errors"
"fmt"
"strconv"
"strings"
"time"
"github.com/nats-io/nats.go"
"github.com/nats-io/nats.go/jetstream"
"github.com/novox/mesh-controller/internal/broker"
)
// A pull request's check is queued by the pull request (novox/hq ADR 0406 §3, §4, issue 974).
// SupersededByKey is the outcome of a check that stopped because its pull request's key moved on: its head or its trunk
// commit is no longer the one it was asked for. Said as the check's failure, as a kill is; nothing is asked of the forge.
const SupersededByKey = "superseded: the pull request's key moved on"
// unsafeToken is what may not stand in a subject token or a key.
var unsafeToken = strings.NewReplacer(".", "_", "*", "_", ">", "_", " ", "_", "/", "_", "\t", "_", "\n", "_")
// tokenOf is a name as one subject token: any dot, wildcard or space replaced by an underscore.
func tokenOf(name string) string {
if name == "" {
return "_"
}
return unsafeToken.Replace(name)
}
// PullKeyOf is a pull request's key in mesh-delivery's bucket, and the last tokens of the subject its check is asked on:
// `<owner>.<repo>.<number>`, a dot in a name made an underscore.
func PullKeyOf(owner, repo string, number int) string {
return tokenOf(owner) + "." + tokenOf(repo) + "." + strconv.Itoa(number)
}
// CheckSubjectOf is the subject a check is asked on, and whether it has one of its own: a check of a pull request on
// the ready queue or the other, by the pull request; a delivery group's composed check by `group.<name>`; any other
// check (a branch tried before its pull request) by its own id, so it replaces nothing.
func CheckSubjectOf(seat string, r BuildRequest) (string, bool) {
c := r.Check
if seat != broker.TheBuildSeat || c == nil {
return "", false
}
prefix := broker.CheckSubjectPrefix(seat, c.Ready)
switch {
case c.Group != "":
return prefix + "group." + tokenOf(c.Group), true
case c.Try || c.Number <= 0:
return prefix + "try." + tokenOf(r.ID), true
}
return prefix + PullKeyOf(c.Owner, c.Repo, c.Number), true
}
// WorkSubjectOf is where an ask is put: a check on the build seat on its own queue, anything else on the seat's.
func WorkSubjectOf(seat string, r BuildRequest) string {
if subject, ok := CheckSubjectOf(seat, r); ok {
return subject
}
return BuildWorkOf(seat)
}
// PullKey is what mesh-delivery keeps of a pull request (ADR 0406 §4): its head, the trunk commit its current check is
// asked against, and its verdict once judged.
type PullKey struct {
Head string `json:"head"`
Trunk string `json:"trunk"`
Verdict string `json:"verdict,omitempty"`
}
// sameCommit says two spellings name one commit; either may be short, but not under twelve characters.
func sameCommit(a, b string) bool {
return a == b || SameHead(a, b)
}
// Moved says a key no longer holds what a check was asked for: its head or its trunk commit differs.
func (k PullKey) Moved(head, trunk string) bool {
return !sameCommit(k.Head, head) || (trunk != "" && k.Trunk != "" && !sameCommit(k.Trunk, trunk))
}
// Judged says a key holds the verdict of the check of this head against this trunk commit.
func (k PullKey) Judged(head, trunk string) bool {
return k.Verdict != "" && sameCommit(k.Head, head) && sameCommit(k.Trunk, trunk) && trunk != ""
}
// ReadPullKey reads a pull request's key from mesh-delivery's bucket: false when it has none.
func ReadPullKey(ctx context.Context, conn *nats.Conn, owner, repo string, number int) (PullKey, bool, error) {
api, err := jetstream.New(conn)
if err != nil {
return PullKey{}, false, err
}
kv, err := api.KeyValue(ctx, broker.PullsBucket)
if err != nil {
return PullKey{}, false, fmt.Errorf("the pull requests' keys: %w", err)
}
e, err := kv.Get(ctx, PullKeyOf(owner, repo, number))
if errors.Is(err, jetstream.ErrKeyNotFound) {
return PullKey{}, false, nil
}
if err != nil {
return PullKey{}, false, err
}
var key PullKey
if err := json.Unmarshal(e.Value(), &key); err != nil {
return PullKey{}, false, fmt.Errorf("the key of %s/%s#%d is not a pull request's: %w", owner, repo, number, err)
}
return key, true, nil
}
// WatchPull calls moved once when the pull request's key moves on from what this check was asked for — its head or its
// trunk commit differs — and not for a key that only gains its verdict. It hears updates, not the key as it stands: the
// key and the request are written for one announcement by two writers, so a key found older than the request is not
// a move. The check is then stopped by whoever asked for it, without asking the forge (novox/hq issue 726). The stop
// ends the watch.
func WatchPull(ctx context.Context, conn *nats.Conn, r BuildRequest, moved func(PullKey)) (stop func(), err error) {
c := r.Check
if c == nil || c.Group != "" || c.Try || c.Number <= 0 {
return func() {}, nil
}
api, err := jetstream.New(conn)
if err != nil {
return nil, err
}
kv, err := api.KeyValue(ctx, broker.PullsBucket)
if err != nil {
return nil, fmt.Errorf("the pull requests' keys: %w", err)
}
watching, end := context.WithCancel(ctx)
w, err := kv.Watch(watching, PullKeyOf(c.Owner, c.Repo, c.Number), jetstream.UpdatesOnly())
if err != nil {
end()
return nil, err
}
go func() {
defer w.Stop()
for {
select {
case <-watching.Done():
return
case e, ok := <-w.Updates():
if !ok {
return
}
if e == nil || e.Operation() != jetstream.KeyValuePut {
continue
}
var key PullKey
if json.Unmarshal(e.Value(), &key) != nil {
continue
}
if key.Moved(r.Ref, c.Trunk) {
moved(key)
return
}
}
}
}()
return end, nil
}
// checkTakeWait is how long a taking loop waits on one queue for an ask that is not there, and takeIdleWait how long it
// waits on the first when all three are empty.
const (
checkTakeWait = 30 * time.Millisecond
takeIdleWait = time.Second
)
internal/link/checks_nats_test.go
+309
View File
@@ -0,0 +1,309 @@
package link
import (
"context"
"encoding/json"
"reflect"
"testing"
"time"
"github.com/nats-io/nats.go/jetstream"
"github.com/novox/mesh-controller/internal/broker"
)
// The checks of pull requests are queued by the pull request (novox/hq ADR 0406 §3, §4, issue 974).
func aPullCheck(id string, number int, head, trunk string, ready bool) BuildRequest {
return BuildRequest{ID: id, Repository: "https://forge.example/novox/a.git", Ref: head,
Check: &CheckRequest{Owner: "novox", Repo: "a", Number: number, Trunk: trunk, Ready: ready}}
}
// The subject a check is asked on says its queue and its pull request: ready or other, `<owner>.<repo>.<number>` with
// a dot in a name made an underscore, `group.<name>` for a group's, the id for a try; a build is on the seat's own queue.
func TestACheckIsAskedOnItsQueueBySubject(t *testing.T) {
for _, c := range []struct {
r BuildRequest
want string
}{
{aPullCheck("b1", 7, "h", "t", false), "mesh.seat.node-build-agent.check.other.novox.a.7"},
{aPullCheck("b2", 7, "h", "t", true), "mesh.seat.node-build-agent.check.ready.novox.a.7"},
{BuildRequest{ID: "b3", Check: &CheckRequest{Owner: "no.vox", Repo: "a.b", Number: 12}}, "mesh.seat.node-build-agent.check.other.no_vox.a_b.12"},
{BuildRequest{ID: "b4", Check: &CheckRequest{Owner: "novox", Repo: "a", Group: "g.1", Ready: true}}, "mesh.seat.node-build-agent.check.ready.group.g_1"},
{BuildRequest{ID: "b5", Check: &CheckRequest{Owner: "novox", Repo: "a", Try: true}}, "mesh.seat.node-build-agent.check.other.try.b5"},
{BuildRequest{ID: "b6", Repository: "/r"}, "mesh.seat.node-build-agent.accept.build"},
{BuildRequest{ID: "b7", Repository: "/r", DryRun: true}, "mesh.seat.node-build-agent.accept.build"},
} {
if got := WorkSubjectOf(TheBuildMachine, c.r); got != c.want {
t.Errorf("%s is asked on %s; want %s", c.r.ID, got, c.want)
}
}
// The retired role has no check queues: its checks wait with its builds.
if got := WorkSubjectOf(TheBuildMachineBefore, aPullCheck("b8", 7, "h", "t", true)); got != "mesh.seat.mesh-build-machine.accept.build" {
t.Errorf("a check of the retired role is asked on %s", got)
}
}
func queueOf(t *testing.T, js *broker.JetStream) Queue {
t.Helper()
q, err := ReadQueue(context.Background(), js, TheBuildMachine)
if err != nil {
t.Fatal(err)
}
return q
}
// A newer request for a pull request replaces the one still waiting, by the stream's rule, and each open pull request is
// left with one queued check (acceptance 5).
func TestANewerCheckRequestReplacesOneStillWaiting(t *testing.T) {
js := aBusWithTheBuildRole(t)
ctx := t.Context()
asks := BuildsOn(js, TheBuildMachine)
for _, r := range []BuildRequest{
aPullCheck("check-7-a", 7, "head-a-0000000", "trunk-0-00000", false),
aPullCheck("check-8-a", 8, "head-b-0000000", "trunk-0-00000", false),
aPullCheck("check-7-b", 7, "head-a-0000000", "trunk-1-00000", false), // a merge moved the trunk
aPullCheck("check-8-b", 8, "head-b-0000000", "trunk-1-00000", false),
aPullCheck("check-7-c", 7, "head-a-0000000", "trunk-2-00000", false), // and a second merge within the minute
} {
if err := asks.Ask(ctx, r); err != nil {
t.Fatal(err)
}
}
var got []string
for _, a := range queueOf(t, js).Asks {
if a.State != AskWaiting {
t.Errorf("%s is %s", a.ID, a.State)
}
got = append(got, a.ID)
}
if want := []string{"check-8-b", "check-7-c"}; !reflect.DeepEqual(got, want) {
t.Fatalf("queued %v; want one check a pull request, the newest: %v", got, want)
}
}
// A delivery build is taken before a check of a ready pull request, that before any other check, and the others in the
// order they were asked.
func TestADeliveryBuildIsTakenBeforeAReadyCheckAndThatBeforeAnyOther(t *testing.T) {
js := aBusWithTheBuildRole(t)
ctx, stop := context.WithCancel(t.Context())
defer stop()
asks := BuildsOn(js, TheBuildMachine)
for _, r := range []BuildRequest{
aPullCheck("other-1", 1, "h1", "t", false),
aPullCheck("other-2", 2, "h2", "t", false),
aPullCheck("ready-3", 3, "h3", "t", true),
{ID: "delivery-4", Repository: "/d"},
aPullCheck("other-5", 5, "h5", "t", false),
aPullCheck("ready-6", 6, "h6", "t", true),
} {
if err := asks.Ask(ctx, r); err != nil {
t.Fatal(err)
}
}
took := make(chan string, 8)
machine := MachineOverNATS(js, "anchor")
defer machine.Close()
go func() {
_ = machine.Take(ctx, func(ctx context.Context, work Build) {
took <- work.Request().ID
_ = work.Announce(ctx, BuildResult{ID: work.Request().ID, On: "anchor", Failed: "no"})
_ = work.Done()
})
}()
var order []string
for len(order) < 6 {
select {
case id := <-took:
order = append(order, id)
case <-time.After(15 * time.Second):
t.Fatalf("taken so far: %v", order)
}
}
if want := []string{"delivery-4", "ready-3", "ready-6", "other-1", "other-2", "other-5"}; !reflect.DeepEqual(order, want) {
t.Fatalf("taken as %v; want %v", order, want)
}
}
// aKeyBucket makes mesh-delivery's bucket of pull requests' keys, which the bus here has not been given by that module.
func aKeyBucket(t *testing.T, js *broker.JetStream) jetstream.KeyValue {
t.Helper()
api, err := jetstream.New(js.Conn())
if err != nil {
t.Fatal(err)
}
kv, err := api.CreateKeyValue(t.Context(), jetstream.KeyValueConfig{Bucket: broker.PullsBucket})
if err != nil {
t.Fatal(err)
}
return kv
}
func putKey(t *testing.T, kv jetstream.KeyValue, number int, k PullKey) {
t.Helper()
body, _ := json.Marshal(k)
if _, err := kv.Put(t.Context(), PullKeyOf("novox", "a", number), body); err != nil {
t.Fatal(err)
}
}
// A request left on the third queue for a head and a trunk commit its key says were judged is acknowledged and dropped when a
// builder takes it; one for a head the key does not say judged is run.
func TestARequestLeftBehindForAJudgedHeadIsDroppedWhenTaken(t *testing.T) {
js := aBusWithTheBuildRole(t)
kv := aKeyBucket(t, js)
ctx, stop := context.WithCancel(t.Context())
defer stop()
putKey(t, kv, 7, PullKey{Head: "head-7-000000000", Trunk: "trunk-1-000000", Verdict: "pass"})
putKey(t, kv, 8, PullKey{Head: "head-8-000000000", Trunk: "trunk-1-000000"}) // asked, not judged
putKey(t, kv, 9, PullKey{Head: "head-9-000000000", Trunk: "trunk-2-000000", Verdict: "pass"}) // judged against another trunk
asks := BuildsOn(js, TheBuildMachine)
for _, r := range []BuildRequest{
aPullCheck("left-7", 7, "head-7-000000000", "trunk-1-000000", false),
aPullCheck("fresh-8", 8, "head-8-000000000", "trunk-1-000000", false),
aPullCheck("moved-9", 9, "head-9-000000000", "trunk-1-000000", false),
} {
if err := asks.Ask(ctx, r); err != nil {
t.Fatal(err)
}
}
took := make(chan string, 8)
machine := MachineOverNATS(js, "anchor")
defer machine.Close()
go func() {
_ = machine.Take(ctx, func(ctx context.Context, work Build) {
took <- work.Request().ID
_ = work.Announce(ctx, BuildResult{ID: work.Request().ID, On: "anchor", Failed: "no"})
_ = work.Done()
})
}()
var ran []string
for len(ran) < 2 {
select {
case id := <-took:
ran = append(ran, id)
case <-time.After(15 * time.Second):
t.Fatalf("run so far: %v", ran)
}
}
if want := []string{"fresh-8", "moved-9"}; !reflect.DeepEqual(ran, want) {
t.Fatalf("run %v; want %v, the judged one dropped", ran, want)
}
time.Sleep(300 * time.Millisecond)
if n := len(queueOf(t, js).Asks); n != 0 {
t.Errorf("%d ask(s) are left in the queues; the dropped one was to be acknowledged", n)
}
}
// A running check stops when its key's head moves or its trunk commit does, not when the key only gains its verdict, and a
// group's check, which a member's merge in the group's order writes nothing to the key of, is not stopped at all.
func TestARunningCheckStopsWhenItsKeysHeadMovesAndAGroupsDoesNot(t *testing.T) {
js := aBusWithTheBuildRole(t)
kv := aKeyBucket(t, js)
ctx, stop := context.WithCancel(t.Context())
defer stop()
moved := make(chan PullKey, 4)
req := aPullCheck("check-7", 7, "head-7-000000000", "trunk-1-000000", false)
end, err := WatchPull(ctx, js.Conn(), req, func(k PullKey) { moved <- k })
if err != nil {
t.Fatal(err)
}
defer end()
putKey(t, kv, 7, PullKey{Head: "head-7-000000000", Trunk: "trunk-1-000000"})
putKey(t, kv, 7, PullKey{Head: "head-7-000000000", Trunk: "trunk-1-000000", Verdict: "pass"})
select {
case k := <-moved:
t.Fatalf("stopped by a key that moved nothing: %+v", k)
case <-time.After(500 * time.Millisecond):
}
putKey(t, kv, 7, PullKey{Head: "head-7-111111111", Trunk: "trunk-1-000000"})
select {
case k := <-moved:
if k.Head != "head-7-111111111" {
t.Errorf("stopped by %+v", k)
}
case <-time.After(5 * time.Second):
t.Fatal("a key whose head moved did not stop the check")
}
// And the trunk commit.
req2 := aPullCheck("check-8", 8, "head-8-000000000", "trunk-1-000000", false)
end2, err := WatchPull(ctx, js.Conn(), req2, func(k PullKey) { moved <- k })
if err != nil {
t.Fatal(err)
}
defer end2()
putKey(t, kv, 8, PullKey{Head: "head-8-000000000", Trunk: "trunk-2-000000"})
select {
case <-moved:
case <-time.After(5 * time.Second):
t.Fatal("a key whose trunk commit moved did not stop the check")
}
// A group's check watches nothing: a sibling's merge in the group's order writes nothing to a key, and a key it does
// have is the pull request's own.
group := BuildRequest{ID: "group-1", Ref: "head-7-000000000",
Check: &CheckRequest{Owner: "novox", Repo: "a", Number: 7, Group: "g1"}}
end3, err := WatchPull(ctx, js.Conn(), group, func(k PullKey) { moved <- k })
if err != nil {
t.Fatal(err)
}
defer end3()
putKey(t, kv, 7, PullKey{Head: "head-7-222222222", Trunk: "trunk-9-000000"})
select {
case k := <-moved:
t.Fatalf("a group's check was stopped by %+v", k)
case <-time.After(500 * time.Millisecond):
}
}
func TestAPullKeyIsMovedByHeadOrTrunkAndJudgedByVerdict(t *testing.T) {
k := PullKey{Head: "aaaaaaaaaaaa1111", Trunk: "tttttttttttt1111", Verdict: "pass"}
if k.Moved("aaaaaaaaaaaa", "tttttttttttt") || !k.Moved("bbbbbbbbbbbbbbbb", "tttttttttttt") || !k.Moved("aaaaaaaaaaaa", "uuuuuuuuuuuuuu") {
t.Error("moved is wrong")
}
if !k.Judged("aaaaaaaaaaaa1111", "tttttttttttt1111") || k.Judged("aaaaaaaaaaaa1111", "") || (PullKey{Head: "a", Trunk: "t"}).Judged("a", "t") {
t.Error("judged is wrong")
}
}
// The queue is read across the three streams in the order they are handed out, each ask says the stream it waits in, and an
// ask is removed from that stream.
func TestTheQueueIsReadAcrossTheThreeStreamsAndAnAskIsDeletedFromItsOwn(t *testing.T) {
js := aBusWithTheBuildRole(t)
ctx := t.Context()
asks := BuildsOn(js, TheBuildMachine)
for _, r := range []BuildRequest{
aPullCheck("other-1", 1, "h1", "t", false),
aPullCheck("ready-2", 2, "h2", "t", true),
{ID: "delivery-3", Repository: "/d"},
} {
if err := asks.Ask(ctx, r); err != nil {
t.Fatal(err)
}
}
q := queueOf(t, js)
var ids, streams []string
for _, a := range q.Asks {
ids, streams = append(ids, a.ID), append(streams, a.Stream)
}
if want := []string{"delivery-3", "ready-2", "other-1"}; !reflect.DeepEqual(ids, want) {
t.Fatalf("read %v; want the order they are handed out, %v", ids, want)
}
if want := []string{"SEAT_NODE_BUILD_AGENT", broker.ChecksReadyStream, broker.ChecksOtherStream}; !reflect.DeepEqual(streams, want) {
t.Fatalf("streams %v; want %v", streams, want)
}
ready, _ := q.Find("ready-2")
if w := WorkerOf(TheBuildMachine, ready); w.Name != broker.ChecksReadyWorker {
t.Errorf("the ready check is handed out by %s", w.Name)
}
if removed, err := DeleteAsk(ctx, js, TheBuildMachine, ready); err != nil || removed != ready.Seq {
t.Fatalf("deleted %d, %v", removed, err)
}
if removed, err := DeleteAsk(ctx, js, TheBuildMachine, ready); err != nil || removed != 0 {
t.Fatalf("deleted again: %d, %v", removed, err)
}
if n := len(queueOf(t, js).Asks); n != 2 {
t.Errorf("%d asks left; want 2", n)
}
}
internal/link/events.go
+3
View File
@@ -264,6 +264,9 @@ type PullUpdated struct {
// check's second layer; MergeCheckSaid says the announcer looked.
MergeCheck bool `json:"merge_check,omitempty"`
MergeCheckSaid bool `json:"merge_check_said,omitempty"`
// Ready says the pull request is marked ready (novox/hq ADR 0406 §3): not the forge's word, mesh-delivery's, said
// to the controller with the head to check again; its check is queued before every other check.
Ready bool `json:"ready,omitempty"`
}
// Checked is a pull request's merge check, judged: what the controller says as `checked`.
internal/link/queue.go
+99 -49
View File
@@ -62,11 +62,12 @@ 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 first took it, a move into order kept (novox/hq issue 558).
// Stored is when the stream took it.
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
// Stream is the work queue holding it: the seat's own, or one of the two check queues (novox/hq ADR 0406 §3). Its
// Seq is the sequence there.
Stream string `json:"stream,omitempty"`
// 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.
@@ -201,62 +202,113 @@ func startedAt(s BuildStart) time.Time {
return t
}
// ReadQueue reads a build seat's work queue: the stream's asks, the worker's position, and what the
// seat's events said about the builds behind it. Through the JetStream API alone (STREAM.INFO,
// CONSUMER.INFO, STREAM.MSG.GET), which only the controller's account reaches.
func ReadQueue(ctx context.Context, js *broker.JetStream, seat string) (Queue, error) {
// queueStream is one stream a seat's asks wait in, and the worker that hands them out.
type queueStream struct {
Worker broker.Consumer
Filter string
}
// queueStreamsOf are the streams a seat's asks wait in, in the order a builder takes from them (novox/hq ADR 0406 §3): the
// seat's own queue, where a delivery build waits, then the checks of pull requests marked ready, then every other check.
// Only the current build role has the check queues; the retired one's checks wait on its own.
func queueStreamsOf(seat string) []queueStream {
worker, _ := broker.HolderConsumerFor("", "", broker.DeclaredSeat{Name: seat, Accepts: []string{"build"}})
streams := []queueStream{{Worker: worker, Filter: BuildWorkOf(seat)}}
if seat == broker.TheBuildSeat {
for _, w := range broker.CheckWorkers() {
streams = append(streams, queueStream{Worker: w, Filter: w.Filters[0]})
}
}
return streams
}
// WorkerOf is the worker that hands out an ask, by the queue it waits in.
func WorkerOf(seat string, ask QueuedAsk) broker.Consumer {
streams := queueStreamsOf(seat)
for _, qs := range streams {
if qs.Worker.Stream == ask.Stream {
return qs.Worker
}
}
return streams[0].Worker
}
// ReadQueue reads a build seat's work queues: the streams' asks, the workers' positions, and what the seat's events
// said about the builds behind them. Through the JetStream API alone (STREAM.INFO, CONSUMER.INFO, STREAM.MSG.GET),
// which only the controller's account reaches. The asks come in the order they are handed out: the seat's own queue's,
// the ready checks', the other checks'.
func ReadQueue(ctx context.Context, js *broker.JetStream, seat string) (Queue, error) {
q := Queue{Seat: seat}
api, err := jetstream.New(js.Conn())
if err != nil {
return q, err
}
stream, err := api.Stream(ctx, worker.Stream)
if err != nil {
return q, fmt.Errorf("the %s work queue (%s) cannot be read: %w", seat, worker.Stream, err)
type read struct {
asks []QueuedAsk
delivered uint64
ackPending int
}
consumer, err := stream.Consumer(ctx, worker.Name)
switch {
case errors.Is(err, jetstream.ErrConsumerNotFound):
// No holder has ever been given a worker: everything in the stream is waiting.
case err != nil:
return q, fmt.Errorf("the worker of %s cannot be read: %w", seat, err)
default:
info, err := consumer.Info(ctx)
var reads []read
for i, qs := range queueStreamsOf(seat) {
stream, err := api.Stream(ctx, qs.Worker.Stream)
if err != nil {
return q, fmt.Errorf("the %s work queue (%s) cannot be read: %w", seat, qs.Worker.Stream, err)
}
var r read
consumer, err := stream.Consumer(ctx, qs.Worker.Name)
switch {
case errors.Is(err, jetstream.ErrConsumerNotFound):
// No holder has ever been given a worker: everything in the stream is waiting.
case err != nil:
return q, fmt.Errorf("the worker of %s cannot be read: %w", seat, err)
default:
info, err := consumer.Info(ctx)
if err != nil {
return q, fmt.Errorf("the worker of %s cannot be read: %w", seat, err)
}
r.delivered, r.ackPending = info.Delivered.Stream, info.NumAckPending
if i == 0 {
q.Delivered, q.MaxDeliver = info.Delivered.Stream, info.Config.MaxDeliver
}
q.AckPending += info.NumAckPending
if q.MaxDeliver == 0 {
q.MaxDeliver = info.Config.MaxDeliver
}
}
q.Delivered = info.Delivered.Stream
q.AckPending = info.NumAckPending
q.MaxDeliver = info.Config.MaxDeliver
// Every ask, by next-by-subject from the first: the stream holds only what is unsettled, so this
// is a handful of reads, never the history.
var seq uint64 = 1
for n := 0; n < 10000; n++ {
msg, err := stream.GetMsg(ctx, seq, jetstream.WithGetMsgSubject(qs.Filter))
if errors.Is(err, jetstream.ErrMsgNotFound) {
break
}
if err != nil {
return q, fmt.Errorf("reading the %s work queue: %w", seat, err)
}
a := askOf(msg)
a.Stream = qs.Worker.Stream
r.asks = append(r.asks, a)
seq = msg.Sequence + 1
}
reads = append(reads, r)
}
// Every ask, by next-by-subject from the first: the stream holds only what is unsettled, so this
// is a handful of reads, never the history.
var asks []QueuedAsk
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 q, fmt.Errorf("reading the %s work queue: %w", seat, err)
}
asks = append(asks, askOf(msg))
seq = msg.Sequence + 1
}
// What the events say, from shortly before the oldest ask behind the worker: its start, if any,
// came after it was asked.
// What the events say, from shortly before the oldest ask behind a worker: its start, if any, came after it was asked.
var since time.Time
for _, a := range asks {
if a.Seq <= q.Delivered && (since.IsZero() || (!a.AskedAt.IsZero() && a.AskedAt.Before(since))) {
since = a.AskedAt
behind := false
for _, r := range reads {
for _, a := range r.asks {
if a.Seq <= r.delivered {
behind = true
if since.IsZero() || (!a.AskedAt.IsZero() && a.AskedAt.Before(since)) {
since = a.AskedAt
}
}
}
}
heard := BuildHeard{Started: map[string]BuildStart{}, Outcomes: map[string]bool{}}
if len(asks) > 0 && q.Delivered > 0 {
if behind {
if since.IsZero() {
since = time.Now().Add(-7 * 24 * time.Hour)
}
@@ -264,17 +316,15 @@ func ReadQueue(ctx context.Context, js *broker.JetStream, seat string) (Queue, e
return q, err
}
}
q.Asks = ClassifyQueue(asks, q.Delivered, q.AckPending, heard)
for _, r := range reads {
q.Asks = append(q.Asks, ClassifyQueue(r.asks, r.delivered, r.ackPending, heard)...)
}
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
@@ -282,7 +332,7 @@ func askOf(msg *jetstream.RawStreamMsg) QueuedAsk {
ask.AskedAt = at
}
} else {
// Nothing read is nothing kept: a half-read request would pass for a delivery (novox/hq issue 558).
// Nothing read is nothing kept: a half-read request would pass for a delivery.
ask.Request = BuildRequest{}
}
return ask
internal/link/queue_kind.go
+50
View File
@@ -0,0 +1,50 @@
package link
import (
"context"
"errors"
"strings"
"github.com/nats-io/nats.go"
"github.com/nats-io/nats.go/jetstream"
"github.com/novox/mesh-controller/internal/broker"
)
// The kinds of ask, as the queue says them. **The kind decides the queue** (novox/hq ADR 0406 §3): a delivery build is on
// the seat's own queue, taken before any check; a check of a pull request marked ready is on the second; every other
// check on the third. Nothing orders the asks any more: a builder takes from the queues in that order, and the check
// queues keep one request a pull request. The reordering of one queue by moving asks to its tail, with its pause of the
// worker, its look every two seconds and its second copies, is gone (issue 558 is answered by the streams).
const (
KindDelivery = "delivery"
KindCheck = "check"
)
// 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
}
// DeleteAsk removes a waiting or dead ask from the queue it is in, at its sequence. Says the sequence it removed, 0 when
// it was not there any more: taken and settled, or replaced by a newer request for its pull request.
func DeleteAsk(_ context.Context, js *broker.JetStream, seat string, ask QueuedAsk) (uint64, error) {
stream := ask.Stream
if stream == "" {
worker, _ := broker.HolderConsumerFor("", "", broker.DeclaredSeat{Name: seat, Accepts: []string{"build"}})
stream = worker.Stream
}
err := js.Context().DeleteMsg(stream, ask.Seq)
switch {
case err == nil:
return ask.Seq, nil
case errors.Is(err, nats.ErrMsgNotFound) || errors.Is(err, jetstream.ErrMsgNotFound) ||
strings.Contains(err.Error(), "no message found"):
return 0, nil
}
return 0, err
}
internal/link/queue_order.go
-433
View File
@@ -1,433 +0,0 @@
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;
// - a branch tried before its pull request goes after everything else, however long it has waited: it is
// never promoted ahead of a delivery, and never ahead of a pull request's check (novox/hq issue 686);
// - 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, tries []QueuedAsk
for _, a := range waiting {
if IsTry(a.Request) {
tries = append(tries, a)
continue
}
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)
byAsking(tries)
return append(append(first, then...), tries...)
}
// 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
}
internal/link/queue_order_test.go
-441
View File
@@ -1,441 +0,0 @@
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)
}
}
internal/link/try_test.go
+2 -20
View File
@@ -2,7 +2,6 @@ package link
import (
"encoding/json"
"reflect"
"strings"
"testing"
"time"
@@ -30,25 +29,8 @@ func TestATrysStartNamesNoCommitTheForgeCouldMatch(t *testing.T) {
}
}
// A try stays behind every delivery and every pull request's check, however long it has waited: the ten
// minutes that put a check ahead of later deliveries never promote a try (novox/hq issue 686).
func TestATryIsNeverHandedOutBeforeADeliveryOrAPullRequestsCheck(t *testing.T) {
now := time.Date(2026, 10, 12, 12, 0, 0, 0, time.UTC)
try := func(seq uint64, id string, asked time.Time) QueuedAsk {
a := anAsk(seq, id, asked, true)
a.Request.Check.Try = true
return a
}
waiting := []QueuedAsk{
try(1, "try-old", now.Add(-40*time.Minute)),
try(2, "try-young", now.Add(-time.Minute)),
anAsk(3, "check-young", now.Add(-2*time.Minute), true),
anAsk(4, "delivery", now.Add(-time.Minute), false),
}
want := []string{"delivery", "check-young", "try-old", "try-young"}
if got := idsOf(QueueOrder(waiting, now)); !reflect.DeepEqual(got, want) {
t.Fatalf("handed out as %v, want %v", got, want)
}
// An id that is a pattern never passes for a build's id.
func TestAnIdThatIsAPatternIsNoBuildId(t *testing.T) {
if !SafeBuildID("build-20261014T000000-1a2b") || SafeBuildID("x.>") || SafeBuildID("*") || SafeBuildID("") {
t.Fatal("an id that is a pattern passes for a build's id")
}