mesh-controller/internal/link/builds_nats.go

package link

import (
	"context"
	"encoding/json"
	"errors"
	"fmt"
	"os"
	"time"

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

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

// The build flow on the bus being built.
//
// **One publish where the old bus needed two** (novox/hq ADR 0121). There, the answer went to a
// reply queue and the announcement to an events exchange, because the two audiences were reached by
// two topologies. Here the outcome is the role's own event: whoever asked matches it by the id their
// request carried, the controller records it, the catalogue places it in the graph. So a build
// machine publishes once and needs permission for nothing but its own role's subjects — no reply
// queue to declare, and no grant over anybody's inbox.

// natsBuilds asks for builds over a connection.
type natsBuilds struct {
	js    *broker.JetStream
	owned bool
	// seat is the build role asked: the one that has a holder (ADR 0190 handover), chosen by the
	// controller from what is assigned, so an ask lands where a machine is pulling.
	seat string
}

// BuildsOverNATS is the asking side on the bus being built, asking the current build role. It dials,
// because the command that asks for a build is a one-shot and holds nothing else.
func BuildsOverNATS(address string) (Builders, error) {
	return BuildsOverNATSOn(address, TheBuildMachine)
}

// BuildsOverNATSOn is the asking side for one named build role — during the handover from the one
// build machine to build agents, the role that has a holder (ADR 0190).
func BuildsOverNATSOn(address, seat string) (Builders, error) {
	js, err := broker.Dial(address)
	if err != nil {
		return nil, fmt.Errorf("cannot reach the bus at %s to ask for a build: %w", address, err)
	}
	return &natsBuilds{js: js, owned: true, seat: seat}, nil
}

// BuildsOn is the asking side on a connection the caller holds, which Close leaves open: the serving
// controller asks on its own rather than dialling one per build (novox/hq issue 327).
func BuildsOn(js *broker.JetStream, seat string) Builders {
	return &natsBuilds{js: js, seat: seat}
}

// role is the seat asked: what the asker was made for, or the current build role for one made
// without saying (a test building the struct by hand).
func (b *natsBuilds) role() string {
	if b.seat == "" {
		return TheBuildMachine
	}
	return b.seat
}

func (b *natsBuilds) Close() {
	if b.owned && b.js != nil {
		b.js.Close()
	}
}

// Ask publishes the work and returns; see Builders.
func (b *natsBuilds) Ask(ctx context.Context, request BuildRequest) error {
	body, err := json.Marshal(request)
	if err != nil {
		return err
	}
	publish, cancel := context.WithTimeout(ctx, 30*time.Second)
	defer cancel()
	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)
	}
	return nil
}

// ErrNotHandedOver is a Submit that failed before the build was handed to the build seat: nothing runs. Any
// other error from Submit came after: the build may still run, and its outcome may still come (novox/hq
// issue 325).
var ErrNotHandedOver = errors.New("the build was not handed over")

func (b *natsBuilds) Submit(ctx context.Context, request BuildRequest,
	wait time.Duration) (BuildResult, error) {

	// Subscribed before the ask, so an outcome cannot arrive before there is anywhere for it to
	// land. Core, not the stream: the asker is waiting now, and the durable copy of this outcome is
	// the same event on EVENTS, which the controller records.
	outcomes, err := b.js.Conn().SubscribeSync(BuildOutcomeOf(b.role()))
	if err != nil {
		return BuildResult{}, fmt.Errorf("%w: cannot listen for a build's outcome: %w", ErrNotHandedOver, err)
	}
	defer func() { _ = outcomes.Unsubscribe() }()
	if err := b.js.Conn().Flush(); err != nil {
		return BuildResult{}, fmt.Errorf("%w: %w", ErrNotHandedOver, err)
	}

	body, err := json.Marshal(request)
	if err != nil {
		return BuildResult{}, fmt.Errorf("%w: %w", ErrNotHandedOver, err)
	}
	// Into the role's work queue and awaited: work the bus never accepted must fail here rather than
	// 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(WorkSubjectOf(b.role(), request), body, nats.Context(publish)); err != nil {
		return BuildResult{}, fmt.Errorf("%w: cannot submit a build: %w", ErrNotHandedOver, err)
	}

	waiting, cancelWait := context.WithTimeout(ctx, wait)
	defer cancelWait()
	for {
		// **A wait that hears a cancel** (novox/hq ADR 0219). A person cancelling an ask records its
		// failure without publishing the role's outcome — that subject is the holders', and the
		// controller may not speak for them — so the waiter also looks, every while, at the cancelled
		// set the cancel wrote first. A kill needs no look: the holder announces it as any outcome.
		slice, endSlice := context.WithTimeout(waiting, cancelLook)
		msg, err := outcomes.NextMsgWithContext(slice)
		endSlice()
		if errors.Is(err, context.DeadlineExceeded) && waiting.Err() == nil {
			if cancelled, _ := IsCancelled(b.js.Conn(), b.role(), request.ID); cancelled {
				return BuildResult{ID: request.ID, Repository: request.Repository, Path: request.Path,
					Ref: request.Ref, Source: request.Source, DryRun: request.DryRun, Failed: CancelledByHand}, nil
			}
			continue
		}
		switch {
		case errors.Is(err, context.DeadlineExceeded):
			return BuildResult{}, waitingFor(wait)
		case errors.Is(err, context.Canceled):
			return BuildResult{}, ctx.Err()
		case err != nil:
			return BuildResult{}, fmt.Errorf("waiting for a build's outcome: %w", err)
		}
		result, mine, err := theOutcomeOf(msg.Data, request.ID)
		if err != nil {
			return BuildResult{}, err
		}
		if mine {
			return result, nil
		}
	}
}

// cancelLook is how often a waiting asker looks whether its ask was cancelled.
var cancelLook = 2 * time.Second

// --- the machine's side ---------------------------------------------------------------------

type natsMachine struct {
	js   *broker.JetStream
	on   string
	seat string
	// 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
	// already running finishes. Nil is never paused.
	Paused func() bool
	// Taking is called once, when this machine is bound to its worker and pulls work: from then on it
	// can take a build, which is what its container's health check reads (novox/hq issue 544). Nil is
	// told nothing.
	Taking func()
	// Claim is asked once a request is fetched, just before it is handed over: false hands it back to the
	// queue at once, for another holder. A holder being drained says so here (novox/hq ADR 0335): it marks
	// the request in hand first and reads its drain after, so an ask of `busy` racing the fetch sees one or
	// the other, never neither. Nil claims every one.
	Claim func() bool
}

// MachineOverNATS takes build work from the current build role.
func MachineOverNATS(js *broker.JetStream, on string) BuildMachine {
	return MachineOverNATSOn(js, on, TheBuildMachine)
}

// MachineOverNATSOn takes build work from the role named — the one this machine's credential claims
// (ADR 0190 handover): its asks come from that seat's worker, and what it says about a build goes
// out as that seat's events, so an outcome is heard where the asker listens.
func MachineOverNATSOn(js *broker.JetStream, on, seat string) BuildMachine {
	return &natsMachine{js: js, on: on, seat: seat}
}

// MachineOverNATSWith is MachineOverNATSOn for a holder that can be paused (novox/hq ADR 0219).
func MachineOverNATSWith(js *broker.JetStream, on, seat string, opts MachineOptions) BuildMachine {
	return &natsMachine{js: js, on: on, seat: seat, opts: opts}
}

// pausedPoll is how often a paused holder looks again whether it was resumed, and pausedFetch how
// long one pull of a holder that can be paused waits.
const (
	pausedPoll  = 2 * time.Second
	pausedFetch = 5 * time.Second
)

func (m *natsMachine) Close() {
	for _, q := range m.subs {
		if q.sub != nil {
			_ = q.sub.Unsubscribe()
		}
	}
}

// Take binds to the role's workers and pulls one request at a time, handing each over.
//
// **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 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)
	}
	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)
	}
	if m.opts.Taking != nil {
		m.opts.Taking()
	}
	var bindAgain time.Time

	for {
		if ctx.Err() != nil {
			return nil
		}
		// **Paused takes nothing new** (novox/hq ADR 0219). Asked before the fetch, never during a
		// build: what this machine already took it finishes, and what it has not taken stays in the
		// queue for another holder — or for this one, resumed.
		if m.opts.Paused != nil && m.opts.Paused() {
			select {
			case <-ctx.Done():
				return nil
			case <-time.After(pausedPoll):
			}
			continue
		}
		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)
		}
		msg, q, err := m.next(ctx)
		switch {
		case ctx.Err() != nil:
			// Ours ended: the machine is being stopped.
			return nil
		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
		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)
				continue
			}
			return fmt.Errorf("the bus stopped delivering build work: %w", err)
		}
		// **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
			}
		}
		// **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)
		// And a check says it is running, to the other builders (ADR 0406 §3): its key put now and at each heartbeat,
		// deleted when it ends, whatever the outcome.
		stopRunning := keepRunning(ctx, m.js.Conn(), request, m.on, heartbeatWhileWorking)
		do(ctx, build)
		stopRunning()
		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.Try || (c.Group == "" && c.Number <= 0) {
		return ""
	}
	if c.Group == "" {
		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)
		} else 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 m.beingChecked(ctx, r)
}

// beingChecked says why a request need not run, or nothing: another builder's check of the same head and trunk commit is
// running, and says so under its pull request's key in the bucket of the checks running (ADR 0406 §3). A key a builder
// stopped renewing is gone within the bucket's age, and the request is taken then. Never the ask's own id: a redelivered ask
// is its own, and a key an older builder wrote that names no head or trunk is no match.
func (m *natsMachine) beingChecked(ctx context.Context, r BuildRequest) string {
	running, ok, err := ReadRunning(ctx, m.js.Conn(), r)
	if err != nil {
		fmt.Fprintf(os.Stderr, "could not read whether %s is being checked, so its check runs: %v\n", r.ID, err)
		return ""
	}
	if ok && running.Build != r.ID && running.Same(r) {
		return fmt.Sprintf("%s is being checked at %.12s on %.12s by %s as %s", r.ID, r.Ref, r.Check.Trunk, running.Builder,
			running.Build)
	}
	return ""
}

type natsBuild struct {
	request BuildRequest
	msg     *nats.Msg
	on      string
	js      *broker.JetStream
	// seat is the role this build was taken from; what the machine says about it is that role's.
	seat string
	seq  int
	// delivery is the how-manyth time the queue handed this ask out (novox/hq issue 594); 0 when the
	// server did not say.
	delivery int
}

// deliveryOf is how many times the queue has handed this message out, this time included.
func deliveryOf(msg *nats.Msg) int {
	meta, err := msg.Metadata()
	if err != nil {
		return 0
	}
	return int(meta.NumDelivered)
}

func (b *natsBuild) Request() BuildRequest { return b.request }

// Announce publishes the outcome as the role's own event, once, for all three audiences.
//
// Into the stream, so a controller that was restarting still records it and a catalogue that was
// down still catches up. The asker is listening on core for the same subject and gets it either way:
// a stream delivers to its durable consumers and the plain subscribers both.
func (b *natsBuild) Announce(ctx context.Context, result BuildResult) error {
	// One body, three readers. Whoever asked matches the id; the controller records it; the catalogue
	// needs to know what was built, which only the manifest says — so the result carries it rather
	// than a second message carrying a second shape.
	//
	// **A failed build names no module**, because it produced no module version and the catalogue
	// would otherwise put something in the graph that was never made. The asker still gets its
	// answer: a failure is the answer.
	if result.Failed == "" && result.Commit != "" {
		result.Module = ModuleOf(result.Manifest)
	}
	// **Which run this is** (novox/hq issue 594): stamped by the transport, which alone knows, so every
	// outcome carries it whatever the machine's code filled in.
	if result.Delivery == 0 {
		result.Delivery = b.delivery
	}
	body, err := json.Marshal(result)
	if err != nil {
		return err
	}
	publish, cancel := context.WithTimeout(ctx, 30*time.Second)
	defer cancel()
	if _, err := b.js.Context().Publish(BuildOutcomeOf(b.seat), body, nats.Context(publish)); err != nil {
		return fmt.Errorf("cannot announce a build's outcome: %w", err)
	}
	return nil
}

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 {
	body, err := json.Marshal(startOf(b.request, b.on, b.delivery, time.Now()))
	if err != nil {
		return err
	}
	// **A start that is a restart says so in the build's log** (novox/hq issue 594): a reader following
	// the build sees why it began again, never a silent second run.
	if said := Restarted(b.delivery); said != "" {
		b.Say("restarted", said)
	}
	publish, cancel := context.WithTimeout(ctx, 30*time.Second)
	defer cancel()
	if _, err := b.js.Context().Publish(BuildStartedOf(b.seat), body, nats.Context(publish)); err != nil {
		return fmt.Errorf("cannot say a build started: %w", err)
	}
	return nil
}

// Say publishes one line under the build's id. Core publish, unawaited: the stream that holds the
// role's events captures it on its way through, and a build must not slow to the pace of an ack
// per line. A line the bus did not take is counted anyway, so the gap is visible to a reader.
func (b *natsBuild) Say(step, message string) {
	b.seq++
	body, err := json.Marshal(BuildLine{
		ID: b.request.ID, Seq: b.seq, At: time.Now().UTC().Format(time.RFC3339Nano),
		Step: step, Message: message,
	})
	if err != nil {
		return
	}
	_ = b.js.Conn().Publish(BuildLogOf(b.seat, b.request.ID), body)
}

func (b *natsBuild) Hold(after time.Duration) error { return b.msg.NakWithDelay(after) }

// endCancelled settles an ask that was cancelled as it was taken: its outcome said as failed, as the
// controller already recorded it, so anybody still waiting on it hears the answer — then
// terminated, so the queue neither keeps nor redelivers it.
func endCancelled(ctx context.Context, b *natsBuild) {
	r := b.request
	fmt.Fprintf(os.Stderr, "%s was cancelled by hand as this machine took it; not built\n", r.ID)
	result := BuildResult{ID: r.ID, Repository: r.Repository, Path: r.Path, Ref: r.Ref, On: b.on,
		Source: r.Source, DryRun: r.DryRun, Failed: CancelledByHand}
	if err := b.Announce(ctx, result); err != nil {
		fmt.Fprintf(os.Stderr, "cannot say %s was cancelled: %v\n", r.ID, err)
	}
	_ = b.msg.Term()
}