mesh-controller/internal/link/holds.go

package link

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

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

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

// Checks wait while a delivery builds (novox/hq ADR 0483, issues 1652, 1687 and 1688).
//
// **The layout other parts read** (ADR 0484 clause 12 has the checker read the same key; a layout shared between modules
// belongs in mesh-sdk, and this one is not there yet):
//
//   - bucket `mesh-controller_plans-building` (broker.PlansBuildingBucket): one key per delivery plan in its build phase, named
//     by the plan's id (`plan-<digits>`), value a PlanBuilding as JSON. Written by the controller (lease holder) alone: put when
//     the plan's first build is asked, put again about every thirty seconds, deleted when the phase ends; the bucket's age
//     (two minutes) removes the key of a controller that stopped. A reader lists the keys and needs no value to decide.
//   - key `let-through` in bucket `mesh-controller_checks-running` (broker.CheckRunningBucket), value a LetThrough as JSON:
//     created by the holder that takes a check which waited LetThroughAfter, refreshed at each heartbeat of that check, deleted
//     when it ends. At most one such check runs at a time on the build seat, because the create fails while the key is there.

// LetThroughAfter is how long a check waits, while a delivery builds, before it is let through (ADR 0483 clause 3).
const LetThroughAfter = 30 * time.Minute

// LetThroughKey is the key of the one check let through, in the bucket of the checks running. It has no dot, so it is no
// pull request's key (`<owner>.<repo>.<number>`) and no group's (`group.<name>`).
const LetThroughKey = "let-through"

// PlanBuilding is the value of a plan's key in the bucket of the plans building.
type PlanBuilding struct {
	Plan  string    `json:"plan"`
	Since time.Time `json:"since"`
	Seen  time.Time `json:"seen"`
}

// LetThrough is the value of the let-through key: the builder that runs the check and when it took it.
type LetThrough struct {
	Builder string    `json:"builder"`
	Since   time.Time `json:"since"`
}

// checkGate is what a taking loop asks of the bus to hold checks while a delivery builds; a variable of the machine so a test
// hands out against fakes.
type checkGate interface {
	// Building is the ids of the plans in their build phase: the keys of the bucket. An error is a bucket that cannot be read.
	Building(ctx context.Context) ([]string, error)
	// OldestWaiting is when the first waiting ask of a check queue was asked, read without taking it.
	OldestWaiting(ctx context.Context, q *takeQueue) (time.Time, bool, error)
	// TakeLetThrough creates the let-through key for a builder: false when it is there (another builder runs a check let through).
	TakeLetThrough(ctx context.Context, builder string) (bool, error)
	// KeepLetThrough puts the key again, and DropLetThrough deletes it.
	KeepLetThrough(ctx context.Context, builder string) error
	DropLetThrough(ctx context.Context) error
}

// busGate is the gate over a connection.
type busGate struct{ conn *nats.Conn }

func (g busGate) Building(ctx context.Context) ([]string, error) {
	return BuildingPlans(ctx, g.conn)
}

// BuildingPlans is the ids of the delivery plans in their build phase, from the controller's bucket. A reader that is not the
// build seat's (the checker, ADR 0484 clause 12) calls this. A bucket that cannot be read is an error; none is none building.
func BuildingPlans(ctx context.Context, conn *nats.Conn) ([]string, error) {
	api, err := jetstream.New(conn)
	if err != nil {
		return nil, err
	}
	kv, err := api.KeyValue(ctx, broker.PlansBuildingBucket)
	if err != nil {
		return nil, fmt.Errorf("the plans building: %w", err)
	}
	keys, err := kv.Keys(ctx)
	if errors.Is(err, jetstream.ErrNoKeysFound) {
		return nil, nil
	}
	if err != nil {
		return nil, fmt.Errorf("the plans building: %w", err)
	}
	return keys, nil
}

// OldestWaiting reads the first ask after the last one the check queue's worker handed out: a direct get on the stream, which
// takes nothing from it and counts against no delivery. The ask's own id says when it was asked.
func (g busGate) OldestWaiting(ctx context.Context, q *takeQueue) (time.Time, bool, error) {
	info, err := jsOf(g.conn).ConsumerInfo(q.worker.Stream, q.worker.Name, nats.Context(ctx))
	if err != nil {
		return time.Time{}, false, fmt.Errorf("the %s worker: %w", q.kind, err)
	}
	body, _ := json.Marshal(map[string]any{"seq": info.Delivered.Stream + 1, "next_by_subj": q.worker.Filters[0]})
	reply, err := g.conn.RequestWithContext(ctx, "$JS.API.DIRECT.GET."+q.worker.Stream, body)
	if err != nil {
		return time.Time{}, false, fmt.Errorf("the first waiting ask of the %s queue: %w", q.kind, err)
	}
	if reply.Header.Get("Status") == "404" || reply.Header.Get("Status") == "408" {
		return time.Time{}, false, nil
	}
	return askedAtOf(reply.Data, reply.Header.Get("Nats-Time-Stamp")), true, nil
}

// askedAtOf is when a request was asked: its id says (NewBuildID), the stream's own time otherwise, and a request that says
// neither is taken as asked now, so it is never let through on a guess.
func askedAtOf(data []byte, stored string) time.Time {
	var r BuildRequest
	if json.Unmarshal(data, &r) == nil {
		if at, ok := BuildAskedAt(r.ID); ok {
			return at
		}
	}
	if at, err := time.Parse(time.RFC3339Nano, stored); err == nil {
		return at
	}
	return time.Now()
}

func jsOf(conn *nats.Conn) nats.JetStreamContext {
	js, _ := conn.JetStream()
	return js
}

func (g busGate) TakeLetThrough(ctx context.Context, builder string) (bool, error) {
	kv, err := runningBucket(ctx, g.conn)
	if err != nil {
		return false, err
	}
	body, _ := json.Marshal(LetThrough{Builder: builder, Since: time.Now().UTC()})
	if _, err := kv.Create(ctx, LetThroughKey, body); err != nil {
		if errors.Is(err, jetstream.ErrKeyExists) {
			return false, nil
		}
		return false, err
	}
	return true, nil
}

func (g busGate) KeepLetThrough(ctx context.Context, builder string) error {
	kv, err := runningBucket(ctx, g.conn)
	if err != nil {
		return err
	}
	body, _ := json.Marshal(LetThrough{Builder: builder, Since: time.Now().UTC()})
	_, err = kv.Put(ctx, LetThroughKey, body)
	return err
}

func (g busGate) DropLetThrough(ctx context.Context) error {
	kv, err := runningBucket(ctx, g.conn)
	if err != nil {
		return err
	}
	if err := kv.Delete(ctx, LetThroughKey); err != nil && !errors.Is(err, jetstream.ErrKeyNotFound) {
		return err
	}
	return nil
}

// CheckHold is what a reader of the queue (the controller's own verb) says of why checks wait: the plans building and the check
// let through now, if one.
type CheckHold struct {
	Plans      []string
	LetThrough *LetThrough
}

// ReadCheckHold reads the plans building and the let-through key. A bucket that cannot be read is an error the caller says.
func ReadCheckHold(ctx context.Context, conn *nats.Conn) (CheckHold, error) {
	var h CheckHold
	plans, err := BuildingPlans(ctx, conn)
	if err != nil {
		return h, err
	}
	h.Plans = plans
	kv, err := runningBucket(ctx, conn)
	if err != nil {
		return h, err
	}
	e, err := kv.Get(ctx, LetThroughKey)
	if errors.Is(err, jetstream.ErrKeyNotFound) {
		return h, nil
	}
	if err != nil {
		return h, err
	}
	var l LetThrough
	if json.Unmarshal(e.Value(), &l) == nil {
		h.LetThrough = &l
	}
	return h, nil
}

// gateOf is the machine's gate: the one a test set, or the bus's.
func (m *natsMachine) gateOf() checkGate {
	if m.gate != nil {
		return m.gate
	}
	return busGate{conn: m.js.Conn()}
}

// sayOnce says a fault of the hold the first time it happens and again once it has recovered and happened anew, so a bucket
// that cannot be read is in the log once and not every second.
func (m *natsMachine) sayOnce(kind string, err error) {
	if m.said == nil {
		m.said = map[string]bool{}
	}
	if err == nil {
		delete(m.said, kind)
		return
	}
	if !m.said[kind] {
		m.said[kind] = true
		fmt.Fprintf(os.Stderr, "%s; read as empty, so no check is held for it: %v\n", kind, err)
	}
}

// building says whether a delivery builds: any key in the bucket of the plans building. **A bucket that cannot be read is said
// and read as empty** (ADR 0483 clause 2): a check wrongly held costs a pull request its turn, a fault that held every check
// would stop the mesh checking.
func (m *natsMachine) building(ctx context.Context) []string {
	plans, err := m.gateOf().Building(ctx)
	m.sayOnce("the bucket of the plans building cannot be read", err)
	if err != nil {
		// empty-on-error: ADR 0483 clause 2 — a bucket that cannot be read is said (sayOnce) and read as empty, because a check wrongly held costs a pull request its turn and a fault that held every check would stop the mesh checking.
		return nil
	}
	return plans
}

// letOneThrough takes the oldest waiting check of the two check queues once it waited LetThroughAfter, by the one let-through
// key: nil when none waited that long, when another holder runs the check let through, or when the check was taken by
// another holder in the moment. The caller owns the key while it returns a message and releases it with dropLetThrough.
func (m *natsMachine) letOneThrough(ctx context.Context) (*nats.Msg, *takeQueue) {
	gate := m.gateOf()
	var oldest *takeQueue
	var oldestAt time.Time
	for _, q := range m.subs[1:] {
		if !q.bound {
			continue
		}
		at, found, err := gate.OldestWaiting(ctx, q)
		m.sayOnce("the first waiting check of the "+q.kind+" queue cannot be read", err)
		if err != nil || !found {
			continue
		}
		if oldest == nil || at.Before(oldestAt) {
			oldest, oldestAt = q, at
		}
	}
	if oldest == nil || time.Since(oldestAt) < m.letThroughAfter() {
		return nil, nil
	}
	created, err := gate.TakeLetThrough(ctx, m.on)
	m.sayOnce("the let-through key cannot be taken", err)
	if err != nil || !created {
		return nil, nil
	}
	trying, end := context.WithTimeout(ctx, checkTakeWait)
	defer end()
	fetched, err := oldest.pull.Fetch(1, nats.Context(trying))
	if len(fetched) == 0 {
		if err != nil && !errors.Is(err, context.DeadlineExceeded) && !errors.Is(err, nats.ErrTimeout) && !errors.Is(err, context.Canceled) {
			m.sayOnce("the "+oldest.kind+" check queue cannot be taken from", err)
		}
		m.dropLetThrough(ctx)
		return nil, nil
	}
	// Another holder took the one that was oldest in the moment between the read and the fetch: what came is a younger check,
	// handed back and the key released, never run during a delivery.
	if askedAtOf(fetched[0].Data, "").Add(m.letThroughAfter()).After(time.Now()) {
		_ = fetched[0].Nak()
		m.dropLetThrough(ctx)
		return nil, nil
	}
	m.letKey = true
	return fetched[0], oldest
}

func (m *natsMachine) letThroughAfter() time.Duration {
	if m.opts.LetThroughAfter > 0 {
		return m.opts.LetThroughAfter
	}
	return LetThroughAfter
}

// dropLetThrough deletes the let-through key this holder created.
func (m *natsMachine) dropLetThrough(ctx context.Context) {
	m.letKey = false
	if err := m.gateOf().DropLetThrough(context.WithoutCancel(ctx)); err != nil {
		fmt.Fprintf(os.Stderr, "could not take the let-through key away; it ages out: %v\n", err)
	}
}

// keepLetThrough refreshes the key at every heartbeat while the check it was created for runs, and returns the end of that: it
// stops the refreshing and deletes the key.
func (m *natsMachine) keepLetThrough(ctx context.Context, every time.Duration) (end func()) {
	done := make(chan struct{})
	finished := make(chan struct{})
	go func() {
		defer close(finished)
		tick := time.NewTicker(every)
		defer tick.Stop()
		for {
			select {
			case <-done:
				return
			case <-tick.C:
				putting, stop := context.WithTimeout(context.WithoutCancel(ctx), 5*time.Second)
				if err := m.gateOf().KeepLetThrough(putting, m.on); err != nil {
					fmt.Fprintf(os.Stderr, "could not keep the let-through key: %v\n", err)
				}
				stop()
			}
		}
	}()
	return func() {
		close(done)
		<-finished
		m.dropLetThrough(ctx)
	}
}

// whyHeld words the plans a check is held for, for the log.
func whyHeld(plans []string) string { return strings.Join(plans, ", ") }