mesh-controller/internal/link/queue_order.go

package link

import (
	"context"
	"errors"
	"fmt"
	"sort"
	"strings"
	"sync"
	"time"

	"github.com/nats-io/nats.go"
	"github.com/nats-io/nats.go/jetstream"

	"github.com/novox/mesh-controller/internal/broker"
)

// The order the build queue hands asks out in (novox/hq issue 558).
//
// **A build a walk waits for goes before a pull request's check.** The queue was first come, first served,
// and a merged change's delivery build waited minutes behind checks while its walk, and every walk after
// it, waited on the build. So:
//
//   - a delivery is handed out before any check;
//   - a check that has waited CheckWaitsAtMost goes before every delivery asked after it, so no check
//     waits for ever behind a stream of deliveries;
//   - within a kind, first come, first served, by when it was asked;
//   - an ask whose kind cannot be told counts as a check: a delivery taken late breaks a budget, a check
//     taken late breaks nothing.
//
// **Kept in the stream, never beside it.** The queue is the seat's work queue, which the server hands out
// in stream order to whichever holder pulls; nothing else holds the order, so a restarted controller
// finds it as it was left. An ask that must go behind another is moved to the tail — deleted from where
// it was and published again, its body unchanged, under the same id — while the worker is paused, so no
// holder takes it in the moment it moves. Only waiting asks move: what a holder took is never touched.

// The kinds of ask, as the queue says them.
const (
	KindDelivery = "delivery"
	KindCheck    = "check"
)

// CheckWaitsAtMost is how long a check waits before it goes ahead of the deliveries asked after it.
var CheckWaitsAtMost = 10 * time.Minute

// KindOf is an ask's kind: a check for a pull request's check — a recheck the merge tool asks included —
// a dry run, or an ask nothing could be read from; a delivery for any other build.
func KindOf(r BuildRequest) string {
	if r.ID == "" || r.Check != nil || r.DryRun {
		return KindCheck
	}
	return KindDelivery
}

// askedOrStored is when an ask was asked, as its id says, or else when the queue took it.
func (a QueuedAsk) askedOrStored() time.Time {
	if !a.AskedAt.IsZero() {
		return a.AskedAt
	}
	return a.Stored
}

// QueueOrder is the order the waiting asks are to be handed out in, now: the deliveries and every check
// that has waited CheckWaitsAtMost, by when each was asked, then the other checks, by when each was asked.
func QueueOrder(waiting []QueuedAsk, now time.Time) []QueuedAsk {
	var first, then []QueuedAsk
	for _, a := range waiting {
		if KindOf(a.Request) == KindCheck && now.Sub(a.askedOrStored()) < CheckWaitsAtMost {
			then = append(then, a)
			continue
		}
		first = append(first, a)
	}
	byAsking := func(s []QueuedAsk) {
		sort.SliceStable(s, func(i, j int) bool {
			if ti, tj := s[i].askedOrStored(), s[j].askedOrStored(); !ti.Equal(tj) {
				return ti.Before(tj)
			}
			return s[i].Seq < s[j].Seq
		})
	}
	byAsking(first)
	byAsking(then)
	return append(first, then...)
}

// toMove is the asks to move to the tail, in this order, for the queue to stand as order says: everything
// after the longest start of order that the queue already holds in that order.
func toMove(queue, order []QueuedAsk) []QueuedAsk {
	at := make(map[uint64]int, len(queue))
	for i, a := range queue {
		at[a.Seq] = i
	}
	kept, last := 0, -1
	for kept < len(order) {
		i, ok := at[order[kept].Seq]
		if !ok || i < last {
			break
		}
		last = i
		kept++
	}
	return order[kept:]
}

// orderPasses keeps one pass at a time in this process: two would each resume the worker the other paused.
var orderPasses sync.Mutex

// pausedFor bounds a pause the pass could not end itself: a controller that stops mid-pass leaves the
// worker paused this long at most.
const pausedFor = 20 * time.Second

// PutInOrder moves the seat's waiting asks into the order QueueOrder says, and says each it moved. A queue
// already in order is read and left alone; the worker is paused only while asks move.
func PutInOrder(ctx context.Context, js *broker.JetStream, seat string) ([]string, error) {
	orderPasses.Lock()
	defer orderPasses.Unlock()
	api, err := jetstream.New(js.Conn())
	if err != nil {
		return nil, err
	}
	worker, _ := broker.HolderConsumerFor("", "", broker.DeclaredSeat{Name: seat, Accepts: []string{"build"}})
	waiting, delivered, err := readWaiting(ctx, api, worker, seat)
	if err != nil || len(toMove(waiting, QueueOrder(waiting, time.Now()))) == 0 {
		return nil, err
	}
	// Paused, then read again: what the worker handed out up to the pause is not waiting any more.
	if delivered >= 0 {
		if _, err := api.PauseConsumer(ctx, worker.Stream, worker.Name, time.Now().Add(pausedFor)); err != nil {
			return nil, fmt.Errorf("the %s worker could not be paused to put its queue in order, so it is left "+
				"first come, first served: %w", seat, err)
		}
		defer func() {
			resuming, cancel := context.WithTimeout(context.WithoutCancel(ctx), 5*time.Second)
			defer cancel()
			if _, err := api.ResumeConsumer(resuming, worker.Stream, worker.Name); err != nil {
				fmt.Printf("the %s worker could not be resumed after its queue was put in order; it resumes by "+
					"itself within %s: %v\n", seat, pausedFor, err)
			}
		}()
		if waiting, _, err = readWaiting(ctx, api, worker, seat); err != nil {
			return nil, err
		}
	}
	var moved []string
	for _, a := range toMove(waiting, QueueOrder(waiting, time.Now())) {
		ok, err := moveToTheTail(ctx, js, api, worker, seat, a.Seq)
		if err != nil {
			return moved, err
		}
		if ok {
			moved = append(moved, a.ID)
		}
	}
	return moved, nil
}

// readWaiting is every ask the worker has not yet handed out, in queue order, and the last sequence it
// handed out; -1 when there is no worker, so nothing is handed out to pause.
func readWaiting(ctx context.Context, api jetstream.JetStream, worker broker.Consumer, seat string) ([]QueuedAsk, int64, error) {
	stream, err := api.Stream(ctx, worker.Stream)
	if err != nil {
		return nil, 0, fmt.Errorf("the %s work queue (%s) cannot be read: %w", seat, worker.Stream, err)
	}
	delivered := int64(-1)
	if consumer, err := stream.Consumer(ctx, worker.Name); err == nil {
		info, err := consumer.Info(ctx)
		if err != nil {
			return nil, 0, fmt.Errorf("the worker of %s cannot be read: %w", seat, err)
		}
		delivered = int64(info.Delivered.Stream)
	} else if !errors.Is(err, jetstream.ErrConsumerNotFound) {
		return nil, 0, fmt.Errorf("the worker of %s cannot be read: %w", seat, err)
	}
	var out []QueuedAsk
	seq := uint64(max(delivered, 0)) + 1
	for n := 0; n < 10000; n++ {
		msg, err := stream.GetMsg(ctx, seq, jetstream.WithGetMsgSubject(BuildWorkOf(seat)))
		if errors.Is(err, jetstream.ErrMsgNotFound) {
			break
		}
		if err != nil {
			return nil, 0, fmt.Errorf("reading the %s work queue: %w", seat, err)
		}
		out = append(out, askOf(msg))
		seq = msg.Sequence + 1
	}
	return out, delivered, nil
}

// moveToTheTail deletes the ask at seq and publishes its body again at the queue's tail. Deleted first, as
// a dead letter delivered again is (novox/hq ADR 0293): one already gone — taken, cancelled, or moved by
// another pass — is not published twice. And looked at once more just before: one the worker has handed
// out since — another process's pass resumed it — is a holder's now and stays. False when it was not moved.
func moveToTheTail(ctx context.Context, js *broker.JetStream, api jetstream.JetStream, worker broker.Consumer, seat string,
	seq uint64) (bool, error) {
	stream, err := api.Stream(ctx, worker.Stream)
	if err != nil {
		return false, err
	}
	if consumer, err := stream.Consumer(ctx, worker.Name); err == nil {
		info, err := consumer.Info(ctx)
		if err != nil {
			return false, err
		}
		if info.Delivered.Stream >= seq {
			return false, nil
		}
	}
	msg, err := stream.GetMsg(ctx, seq)
	if errors.Is(err, jetstream.ErrMsgNotFound) {
		return false, nil
	}
	if err != nil {
		return false, err
	}
	if err := stream.DeleteMsg(ctx, seq); err != nil {
		if errors.Is(err, jetstream.ErrMsgNotFound) || errors.Is(err, jetstream.ErrMsgDeleteUnsuccessful) {
			return false, nil
		}
		return false, err
	}
	again := &nats.Msg{Subject: msg.Subject, Data: msg.Data, Header: nats.Header{}}
	for k, v := range msg.Header {
		// The server's own headers stay behind: a message id would be taken for a duplicate and dropped.
		if !strings.HasPrefix(k, "Nats-") {
			again.Header[k] = v
		}
	}
	for try := 0; ; try++ {
		publishing, cancel := context.WithTimeout(ctx, 10*time.Second)
		_, err := js.Context().PublishMsg(again, nats.Context(publishing))
		cancel()
		if err == nil {
			return true, nil
		}
		if try == 2 {
			return false, fmt.Errorf("the ask at seq %d of the %s queue was taken out to be put behind and could "+
				"NOT be put back: %w — its asker hears no outcome and must ask again", seq, seat, err)
		}
		time.Sleep(time.Second)
	}
}