mesh-delivery/modules/mesh-delivery-builder/internal/build/askagain.go

package build

import (
	"context"
	"errors"
	"fmt"
	"regexp"
	"slices"
	"sort"
	"strconv"
	"strings"
	"time"

	"git.novox.be/novox/mesh-delivery/modules/mesh-delivery-builder/buildrun"
	"git.novox.be/novox/mesh-delivery/modules/mesh-delivery-builder/builds"
	sdkstep "git.novox.be/novox/mesh-delivery/sdk/buildstep"
)

// **An ask that can no longer be answered is asked again** (novox/hq issue 1945). A build seat's stream gives an ask up
// once it was delivered as often as its worker may (sdkstep.AskMaxDeliver), and a message given up on, dropped or
// terminated with no step written is answered by no build node: the lane waiting for its step would wait for ever, and on
// restart take over the same id and wait again. The builder notices it at the check bound (CheckAfter, the step bound),
// a context's bound and no timer of its own: it asks Gone, which reads the bus, and asks such an ask again under a new
// id, written into the key, said in its log and as the event build.asked-again, at most Askings times; then the key is
// failed with why. An ask DEAD_LETTERS says the seat gave up (buildrun.LostAskWaiter, issue 734) is asked again at once.
// A person asks the same by hand with the tool builder_ask_again (AskAgain).

// DefaultAskings is how many times a build module's part of a build is asked, the first included, when Builder.Askings is 0.
const DefaultAskings = 3

// AskedAgainEvent is the event the builder says when it asked an ask again: mesh.mod.mesh-delivery-builder.event.<this>.
const AskedAgainEvent = "build.asked-again"

// AskedAgain is the body of AskedAgainEvent.
type AskedAgain struct {
	Key         string    `json:"key"`
	Module      string    `json:"module"`
	FP          string    `json:"fp"`
	BuildModule string    `json:"build_module"`
	Language    string    `json:"language"`
	Old         string    `json:"old"`
	New         string    `json:"new"`
	Asking      int       `json:"asking"`
	Askings     int       `json:"askings"`
	Why         string    `json:"why"`
	At          time.Time `json:"at"`
}

// AskSeen is what the bus says of an ask whose step has not ended, as the builder may read it: its build seat's stream
// (information and messages by sequence) and its build module's steps. The builder may not read the seat workers'
// delivery counts or ack floors (granted to the delivery view alone: $JS.API.CONSUMER.INFO.<stream>.<worker>), so a
// message given up on is told by what the workers did after it: a build node took an ask asked later, of the same kind or
// of one pulled after it (issue 1945 note 62: seven messages SEAT_NODE_BUILD_GO had given up on sat ahead of a Go ask for
// five hours, each counted as waiting, so the ask was never asked again).
type AskSeen struct {
	Stream string
	// Kind is the ask's kind, At when it was asked (its id's time).
	Kind string
	At   time.Time
	// Step is whether a build node wrote a step for the ask (running: the step bound of buildrun.Asker judges it).
	Step bool
	// OnStream is whether the stream still holds the ask's message; PutAt is when it was put.
	OnStream bool
	PutAt    time.Time
	// Ahead are the messages of its subject before it on the stream.
	Ahead []Queued
	// Last is the ask asked last that a build node took (wrote a step for) of the ask's kind or of a kind pulled after
	// it, when one was taken: every message of the kind asked before it was then delivered already, or given up on.
	Last *Taken
}

// Queued is one message of a build seat's stream: its sequence, its ask's id and kind, when the ask was asked (its id's
// time, the put time when the id says none) and put, and the state of the step a build node wrote for it ("" for none).
type Queued struct {
	Seq   uint64    `json:"seq"`
	Ask   string    `json:"ask"`
	Kind  string    `json:"kind"`
	At    time.Time `json:"asked"`
	PutAt time.Time `json:"put"`
	Step  string    `json:"step,omitempty"`
}

// Taken is an ask a build node took: its id, kind and when it was asked, and the step's node and start.
type Taken struct {
	Ask     string    `json:"ask"`
	Kind    string    `json:"kind"`
	At      time.Time `json:"asked"`
	Node    string    `json:"node"`
	Started time.Time `json:"started"`
}

// Proves is whether a taken ask of kind taken shows that a message of kind waiting asked before it is no longer
// delivered: a holder pulls the kinds in sdkstep.AskKinds' order, so taking one of the same kind or of a later one means
// its worker had none of waiting's kind left to give.
func Proves(taken, waiting string) bool {
	t, w := slices.Index(sdkstep.AskKinds, taken), slices.Index(sdkstep.AskKinds, waiting)
	return t >= 0 && w >= 0 && t >= w
}

// Waits is whether a message ahead still waits to be delivered: no build node wrote a step for it, and no ask of its
// kind asked after it was taken. A message its worker gave up on (delivered its most times) stays on a work queue until
// its age, and is not counted.
func (s AskSeen) Waits(q Queued) bool {
	return q.Step == "" && !s.passed(q.At)
}

func (s AskSeen) passed(at time.Time) bool {
	return s.Last != nil && Proves(s.Last.Kind, s.Kind) && s.Last.At.After(at)
}

// Waiting is how many messages ahead still wait to be delivered (Waits).
func (s AskSeen) Waiting() int {
	n := 0
	for _, q := range s.Ahead {
		if s.Waits(q) {
			n++
		}
	}
	return n
}

// Why is why the ask can no longer be answered, or "" when it can: it has a step, waits behind others of its kind still
// deliverable, or no build node took a later ask of its kind (the nodes are busy or paused: a backlog, never a fault,
// ADR 0525 §9).
func (s AskSeen) Why() string {
	switch {
	case s.Step:
		return ""
	case !s.OnStream:
		return fmt.Sprintf("no build node wrote a step for it and %s no longer holds it: it was given up on or dropped", s.Stream)
	case s.Waiting() > 0 || !s.passed(s.at()):
		return ""
	default:
		behind := ""
		if n := len(s.Ahead); n > 0 {
			behind = fmt.Sprintf(" behind %d given up on or taken", n)
		}
		return fmt.Sprintf("it waits on %s since %s%s while %s took the %s ask %s asked later at %s: its stream no longer delivers it (given up after its most deliveries)",
			s.Stream, s.PutAt.UTC().Format(time.RFC3339), behind, s.Last.Node, s.Last.Kind, s.Last.Ask, s.Last.At.UTC().Format(time.RFC3339))
	}
}

func (s AskSeen) at() time.Time {
	if s.At.IsZero() {
		return s.PutAt
	}
	return s.At
}

// AskedAt is when an ask id was made (sdkstep.NewID: its milliseconds in base 36 before six hex digits), or zero.
func AskedAt(id string) time.Time {
	if len(id) <= 6 {
		return time.Time{}
	}
	ms, err := strconv.ParseInt(id[:len(id)-6], 36, 64)
	if err != nil || ms <= 0 {
		return time.Time{}
	}
	return time.UnixMilli(ms)
}

// Dropper removes an ask's message from its build seat's stream (Builder.Gone may be one): the old message of an ask asked
// again, so a message given up on does not sit ahead of every later ask of its kind until its age (issue 1945 note 62).
type Dropper interface {
	Drop(ctx context.Context, a sdkstep.StepAsk) error
}

// drop removes the old ask's message when Gone can; a refusal is said and changes nothing else: the new ask is put.
func (b *Builder) drop(ctx context.Context, ask sdkstep.StepAsk, old string) {
	d, ok := b.Gone.(Dropper)
	if !ok {
		return
	}
	prior := ask
	prior.ID = old
	if err := d.Drop(ctx, prior); err != nil {
		b.logf("the message of the ask %s asked again as %s stays on its stream: %v", old, ask.ID, err)
	}
}

// DefaultDeclineBound is how long one ask may go on being declined, from its first decline, before its key fails (the
// checker's DefaultDeclineBound, issue 890); DefaultDeclinePause is how long the builder waits before it puts a declined
// ask again, so a lone node off the agreed toolchain cannot spin on it.
const (
	DefaultDeclineBound = 30 * time.Minute
	DefaultDeclinePause = 5 * time.Second
)

func (b *Builder) declineBound() time.Duration {
	if b.DeclineBound > 0 {
		return b.DeclineBound
	}
	return DefaultDeclineBound
}

func (b *Builder) declinePause() time.Duration {
	if b.DeclinePause != 0 {
		return max(b.DeclinePause, 0)
	}
	return DefaultDeclinePause
}

// declinedWords names the nodes that declined an ask and why, in the order of their names.
func declinedWords(by map[string]string) string {
	nodes := make([]string, 0, len(by))
	for n := range by {
		nodes = append(nodes, n)
	}
	sort.Strings(nodes)
	var parts []string
	for _, n := range nodes {
		parts = append(parts, n+": "+by[n])
	}
	return strings.Join(parts, "; ")
}

func (b *Builder) askings() int {
	if b.Askings > 0 {
		return b.Askings
	}
	return DefaultAskings
}

var askingWords = regexp.MustCompile(`asked (\S+) again \(asking (\d+)\)`)

// askingOf is the asking a key's step words name for a build module's ask, 1 when none.
func askingOf(step, buildModule string) int {
	for _, m := range askingWords.FindAllStringSubmatch(step, -1) {
		if m[1] == buildModule {
			if n, err := strconv.Atoi(m[2]); err == nil && n > 0 {
				return n
			}
		}
	}
	return 1
}

// AskAgain asks the waiting asks of a build key again now, under new ids (the tool builder_ask_again): it answers the
// languages whose lane waits, none when no lane of this builder waits on the key.
func (b *Builder) AskAgain(key, why string) []string {
	b.init()
	b.mu.Lock()
	defer b.mu.Unlock()
	var asked []string
	for language, c := range b.waiting[key] {
		select {
		case c <- why:
		default: // one is already asked
		}
		asked = append(asked, language)
	}
	return asked
}

// Waiting are the build keys whose lanes wait for a step now, each with its languages.
func (b *Builder) Waiting() map[string][]string {
	b.init()
	b.mu.Lock()
	defer b.mu.Unlock()
	out := map[string][]string{}
	for key, ls := range b.waiting {
		for l := range ls {
			out[key] = append(out[key], l)
		}
	}
	return out
}

func (b *Builder) listen(key, language string) chan string {
	b.init()
	c := make(chan string, 1)
	b.mu.Lock()
	if b.waiting[key] == nil {
		b.waiting[key] = map[string]chan string{}
	}
	b.waiting[key][language] = c
	b.mu.Unlock()
	return c
}

func (b *Builder) unlisten(key, language string, c chan string) {
	b.mu.Lock()
	if b.waiting[key][language] == c {
		delete(b.waiting[key], language)
		if len(b.waiting[key]) == 0 {
			delete(b.waiting, key)
		}
	}
	b.mu.Unlock()
}

var (
	errCheck = errors.New("the check bound passed")
	errAgain = errors.New("asked again")
	// errDeclined: a build node on another toolchain than the agreed one sat out of the ask (ADR 0584 clause 2).
	errDeclined = errors.New("declined")
)

type byHand struct{ why string }

func (h byHand) Error() string { return "asked again by hand: " + h.why }

// answer waits for the step of the lane's ask, and asks it again while it can no longer be answered (Gone) or a person
// asked it again (AskAgain); it answers the step and the ask that has it.
func (b *Builder) answer(ctx context.Context, l *lane, ask sdkstep.StepAsk, asked *askNotes, module, fp string) (sdkstep.Step, sdkstep.StepAsk, error) {
	key, _ := builds.BuilderBuildKey(module, fp)
	again := b.listen(key, ask.Language)
	defer b.unlisten(key, ask.Language, again)
	asking := askingOf(asked.step, l.module)
	var firstDecline time.Time
	declines := map[string]string{} // node -> why it declined, the last
	for {
		step, why, err := b.waitOnce(ctx, ask, again)
		switch {
		case errors.Is(err, errCheck):
			continue
		case errors.Is(err, errDeclined):
			// Not an asking: a node that sits out would use them all while it idles and pulls the ask again at once.
			// Bounded by time from the first decline, then failed loudly naming the nodes and their reasons (as the
			// checker's decline loop is, issue 890).
			now := b.now()
			if firstDecline.IsZero() {
				firstDecline = now
			}
			declines[step.Node] = why
			if bound := b.declineBound(); now.Sub(firstDecline) > bound {
				return step, ask, fmt.Errorf("the ask %s of %s was declined for %s and no build node took it: %s", ask.ID, l.module, bound.Round(time.Second), declinedWords(declines))
			}
			// The pause is a context's bound, not a timer (ADR 0437 §8).
			pause, stop := context.WithTimeout(ctx, b.declinePause())
			<-pause.Done()
			stop()
			if ctx.Err() != nil {
				return step, ask, ctx.Err()
			}
			old := ask.ID
			ask.ID, ask.At = sdkstep.NewID(), time.Now()
			if err := asked.note(ctx, ask.Language, ask.ID, fmt.Sprintf("put %s again after %s declined it", l.module, step.Node)); err != nil {
				return step, ask, fmt.Errorf("the ask of %s put again after a decline could not be written to the key: %w", l.module, err)
			}
			if err := b.Asks.Put(ctx, ask); err != nil {
				return step, ask, err
			}
			b.drop(ctx, ask, old)
			b.logf("%s: the ask %s of %s was declined on %s and is put again as %s: %s", key, old, l.module, step.Node, ask.ID, why)
		case errors.Is(err, errAgain):
			if asking >= b.askings() {
				return step, ask, fmt.Errorf("the ask %s could not be answered and was asked %d times, the most it is: %s", ask.ID, asking, why)
			}
			asking++
			old := ask.ID
			ask.ID, ask.At = sdkstep.NewID(), time.Now()
			if err := asked.note(ctx, ask.Language, ask.ID, fmt.Sprintf("asked %s again (asking %d): %s", l.module, asking, why)); err != nil {
				return step, ask, fmt.Errorf("the ask of %s asked again could not be written to the key: %w", l.module, err)
			}
			if err := b.Asks.Put(ctx, ask); err != nil {
				return step, ask, err
			}
			b.drop(ctx, ask, old)
			b.logf("%s: the ask %s of %s is asked again as %s (asking %d of %d): %s", key, old, l.module, ask.ID, asking, b.askings(), why)
			if b.Said != nil {
				b.Said(AskedAgain{Key: key, Module: module, FP: fp, BuildModule: l.module, Language: ask.Language, Old: old, New: ask.ID,
					Asking: asking, Askings: b.askings(), Why: why, At: ask.At})
			}
		default:
			return step, ask, err
		}
	}
}

// waitOnce waits for the ask's step until it ends, a person asks it again, or the check bound passes: then it asks Gone
// and answers errAgain with why when the ask can no longer be answered, errCheck when it waits on.
func (b *Builder) waitOnce(ctx context.Context, ask sdkstep.StepAsk, again <-chan string) (sdkstep.Step, string, error) {
	hand, cancel := context.WithCancelCause(ctx)
	defer cancel(nil)
	go func() {
		select {
		case why := <-again:
			cancel(byHand{why})
		case <-hand.Done():
		}
	}()
	wait := hand
	if b.Gone != nil && b.CheckAfter > 0 {
		var stop context.CancelFunc
		wait, stop = context.WithTimeoutCause(hand, b.CheckAfter, errCheck)
		defer stop()
	}
	step, err := b.Asks.Wait(wait, ask)
	if _, declined := sdkstep.DeclinedOf(step); err == nil && declined {
		// A build node off the agreed toolchain sat out (ADR 0584 clause 2). It is no asking: answer() puts the ask again
		// after a pause, within DeclineBound.
		return step, step.Reason, errDeclined
	}
	if err == nil && buildrun.IsLost(step) {
		// The build seat gave the ask up (issue 734): it can no longer be answered, so it is asked again now, with the
		// words of why, not at the check bound.
		return step, step.Reason, errAgain
	}
	if err == nil || ctx.Err() != nil {
		return step, "", err
	}
	var h byHand
	if errors.As(context.Cause(hand), &h) {
		return step, strings.TrimSpace("asked again by hand: " + h.why), errAgain
	}
	if errors.Is(context.Cause(wait), errCheck) {
		why, gerr := b.Gone.Gone(ctx, ask)
		if gerr != nil {
			b.logf("whether the ask %s can still be answered cannot be read, so it is waited for: %v", ask.ID, gerr)
			return step, "", errCheck
		}
		if why == "" {
			return step, "", errCheck
		}
		return step, why, errAgain
	}
	return step, "", err
}