mesh-delivery/modules/mesh-delivery-builder/buildrun/asks.go

package buildrun

import (
	"context"
	"encoding/json"
	"errors"
	"fmt"
	"log"
	"strings"
	"sync"
	"time"

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

	"git.novox.be/novox/mesh-delivery/sdk/buildstep"
	"git.novox.be/novox/mesh-sdk/go/busretry"
)

// **An asker puts an ask on its build seat's stream and waits for its step** (novox/hq ADR 0525 §1, §6; issue 1887 A4,
// A5). It chooses no node: whichever build node pulls the ask writes `<node>.<ask id>` in its build module's steps state,
// and the asker reads the first such key that is not running. The builder and the checker share it.

// AskPublisher puts an encoded ask on subject, de-duplicated by its id.
type AskPublisher interface {
	PublishAsk(ctx context.Context, subject, id string, data []byte) error
}

// AskWaiter watches an ask's steps: every value a key `*.<ask id>` of the build module's steps state takes, running or
// ended, on the channel it answers, until ctx ends. A watch, not a poll.
type AskWaiter interface {
	WatchAsk(ctx context.Context, module, askID string) (<-chan buildstep.Step, error)
}

// Asker puts asks and waits for their steps.
type Asker struct {
	Publisher AskPublisher
	Waiter    AskWaiter
	// Seats finds the build module of an ask's language (design 47 G A-82).
	Seats *BuildSeats
	// Bound is how long a running step (a build node took the ask, its key written running) may stand still, with no new phase,
	// before the asker stops waiting; a step that keeps moving is waited for up to 3 Bounds from its start (issue 1930); zero waits as long as ctx. An ask waiting in its stream is not bounded: the stream's backlog is no fault
	// (ADR 0525 §6, §9), and the bound starts again when another node takes the ask.
	Bound time.Duration
	// Dropper, when set, is told to end an ask the asker stops waiting for without its end (ADR 0664 §5): a step given
	// again that the bus did not give within its bound, or a wait ended with a cause its caller names (a superseded
	// fingerprint). Never a wait ended by a shutdown (a plain cancel): the ask is then taken over by the next run.
	Dropper AskDropper
	// Bounds and Expiry keep the step bounds as keys with their own time to live (design 47 G A-77, A-101, A-118): the stand-still
	// bound as StandBoundKey and the cap of three Bounds as CapBoundKey in the asking part's own state, whose expiry the part's watch
	// of that state tells Expiry (Expiry.Ended), and Wait then ends. Nil Bounds keeps the bounds as timers of the process, as the
	// checker has them.
	Bounds StepBounds
	Expiry *Expiry
	// Logf says a drop that failed (the log's Printf when unset): the ask then stays on its stream, and it is said.
	Logf func(format string, args ...any)
}

// StepBounds keeps one bound as a key with its own time to live: Arm writes key afresh to end after bound, End deletes it.
type StepBounds interface {
	Arm(ctx context.Context, key, why string, bound time.Duration) error
	End(ctx context.Context, key string)
}

// The keys of an ask's step bounds in the asking part's state: `stand.<ask id>` (the step stood still) and `cap.<ask id>` (a step
// that kept moving never ended).
const (
	standBoundPrefix = "stand."
	capBoundPrefix   = "cap."
)

// StandBoundKey is the key of the stand-still bound of an ask's step.
func StandBoundKey(askID string) string { return standBoundPrefix + askID }

// CapBoundKey is the key of the cap on an ask's step.
func CapBoundKey(askID string) string { return capBoundPrefix + askID }

// StepBoundOf says whether key is a step bound's, as the part's watch of its bounds state reads it.
func StepBoundOf(key string) bool {
	return strings.HasPrefix(key, standBoundPrefix) || strings.HasPrefix(key, capBoundPrefix)
}

// Expiry is where the part's watch of its bounds state tells the end of a bound key (the bus's marker of its time to live), and the
// waits for those keys hear it. Safe for concurrent use; the zero value is ready.
type Expiry struct {
	mu    sync.Mutex
	waits map[string]chan struct{}
}

// Ended tells the end of key to the wait that holds it, and says whether one did.
func (e *Expiry) Ended(key string) bool {
	e.mu.Lock()
	c, ok := e.waits[key]
	e.mu.Unlock()
	if ok {
		select {
		case c <- struct{}{}:
		default:
		}
	}
	return ok
}

// hold is the channel key's end is told on, and its release.
func (e *Expiry) hold(key string) (<-chan struct{}, func()) {
	e.mu.Lock()
	defer e.mu.Unlock()
	if e.waits == nil {
		e.waits = map[string]chan struct{}{}
	}
	c := make(chan struct{}, 1)
	e.waits[key] = c
	return c, func() {
		e.mu.Lock()
		if e.waits[key] == c {
			delete(e.waits, key)
		}
		e.mu.Unlock()
	}
}

// Withdrawn is the cause a caller gives the context of an ask's wait to withdraw the ask (ADR 0664 §4): a build superseded by a
// newer fingerprint, a candidate taken out, an ask asked again by hand. Asker.Wait drops the ask on this cause alone: a
// plain cancel (a shutdown) and any other cause (a bound passing, a check of whether the ask is gone) leave it on its stream
// for the next run. A cause that wraps a Withdrawn (errors.As) withdraws too.
type Withdrawn struct{ Why string }

func (w Withdrawn) Error() string { return w.Why }

// AskDropper ends an ask: the seat's verb drop, served by any holder of the build seat (no node is named). It writes the
// ask in the build module's shared state of pauses, so every node refuses it when the stream gives it, and the node that
// runs it ends the step (ADR 0664 §5).
type AskDropper interface {
	DropAsk(ctx context.Context, a buildstep.StepAsk, why string) error
}

// SeatCaller calls a seat's verb on whichever holder answers (mesh-sdk process.Conn.Call).
type SeatCaller interface {
	Call(key string, body any) (json.RawMessage, error)
}

// SeatDrops is an AskDropper that calls the verb drop of the ask's build seat over the bus.
type SeatDrops struct{ Caller SeatCaller }

// DropAsk calls <seat>.drop for a, naming its asker and why.
func (d SeatDrops) DropAsk(_ context.Context, a buildstep.StepAsk, why string) error {
	seat, err := buildstep.BuildSeat(a.Language)
	if err != nil {
		return err
	}
	if _, err := d.Caller.Call("seat:"+seat+".drop", map[string]any{"id": a.ID, "by": a.Asker, "why": why}); err != nil {
		return fmt.Errorf("dropping the ask %s: %w", a.ID, err)
	}
	return nil
}

// dropped ends a without its end. A failed drop is not the wait's error, but it is said (Logf).
func (k Asker) dropped(a buildstep.StepAsk, why string) {
	if k.Dropper == nil {
		return
	}
	ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
	defer cancel()
	if err := k.Dropper.DropAsk(ctx, a, why); err != nil {
		say := k.Logf
		if say == nil {
			say = log.Printf
		}
		say("the ask %s was not dropped (%s), it stays on its stream: %v", a.ID, why, err)
	}
}

// PutTry is how long one publish of an ask waits for its stream's answer before it is made again (Put).
var PutTry = 10 * time.Second

// Put puts a on its stream: the seat's accept subject of its language and kind, under the builder's name as asker (design 47 G
// §481, §600: `accept.<queue>.builder`). The put is a kept call (novox/hq ADR 0668
// §1, §2; issue 1966, issue 1748 note 27): it carries the ask's id as its message id, so the stream takes it once, and it
// is made again through every transport fault (a deadline, no responders) within busretry.Bound, each try bounded by
// PutTry, until ctx ends. A put that still meets one is an error opening `transport fault:` (TransportFault), so the key it
// fails is one the builder does not take again by itself (ADR 0664 §3) and builder_build_again does.
func (k Asker) Put(ctx context.Context, a buildstep.StepAsk) error {
	raw, err := buildstep.EncodeStepAsk(a)
	if err != nil {
		return err
	}
	subject, err := buildstep.AskSubjectBy(a.Language, a.Kind, buildstep.AskerBuilder)
	if err != nil {
		return err
	}
	_, err = busretry.DoRepeatable("putting the "+a.Kind+" ask "+a.ID, func() (struct{}, error) {
		if err := ctx.Err(); err != nil {
			return struct{}{}, fmt.Errorf("the put was given up: %v", err) // not a fault to wait out: the build ended
		}
		try, cancel := context.WithTimeout(ctx, PutTry)
		defer cancel()
		err := k.Publisher.PublishAsk(try, subject, a.ID, raw)
		if err != nil && ctx.Err() != nil {
			return struct{}{}, fmt.Errorf("the put was given up: %v", err)
		}
		return struct{}{}, err
	}, nil)
	switch {
	case err == nil:
		return nil
	case busretry.TimedOut(err) || busretry.Undelivered(err):
		seat, _ := buildstep.BuildSeat(a.Language)
		return errors.New(TransportFault(FaultClass(err.Error()), seat, "no node", fmt.Sprintf("the put of the %s ask %s: %v", a.Kind, a.ID, err)))
	}
	return fmt.Errorf("putting the %s ask %s: %w", a.Kind, a.ID, err)
}

// Wait waits for the step of an ask already put (a builder that started again takes over its own asks with it, never
// putting them twice) and answers it when it ended, whatever its state. The bound runs from the step's start on the node
// that took the ask, read from its running key, so an ask that waits in the stream for a free slot waits as long as ctx.
func (k Asker) Wait(ctx context.Context, a buildstep.StepAsk) (buildstep.Step, error) {
	bm, err := k.Seats.OfAskLanguage(a.Language)
	if err != nil {
		return buildstep.Step{}, err
	}
	module := bm.Module
	watch, cancel := context.WithCancel(ctx)
	defer cancel()
	steps, err := k.Waiter.WatchAsk(watch, module, a.ID)
	if err != nil {
		return buildstep.Step{}, err
	}
	// The two bounds: keys with their own time to live whose expiry is told by Expiry, or, with no Bounds, timers.
	var bound, outerCap <-chan struct{}
	keyed := k.Bounds != nil && k.Expiry != nil
	var timers [2]*time.Timer
	standKey, capKey := StandBoundKey(a.ID), CapBoundKey(a.ID)
	if keyed {
		var releaseStand, releaseCap func()
		bound, releaseStand = k.Expiry.hold(standKey)
		outerCap, releaseCap = k.Expiry.hold(capKey)
		defer func() {
			releaseStand()
			releaseCap()
			k.Bounds.End(context.WithoutCancel(ctx), standKey)
			k.Bounds.End(context.WithoutCancel(ctx), capKey)
		}()
	}
	defer func() {
		for _, t := range timers {
			if t != nil {
				t.Stop()
			}
		}
	}()
	// arm ends the bound i (0 stand-still, 1 cap) after d: a key's time to live, or a timer.
	arm := func(i int, d time.Duration) error {
		d = max(d, time.Millisecond)
		if keyed {
			key, why := standKey, "the step of the ask "+a.ID+" stood still"
			if i == 1 {
				key, why = capKey, "the step of the ask "+a.ID+" never ended"
			}
			return k.Bounds.Arm(ctx, key, why, d)
		}
		if timers[i] != nil {
			timers[i].Stop()
		}
		c := make(chan struct{}, 1)
		timers[i] = time.AfterFunc(d, func() { c <- struct{}{} })
		if i == 0 {
			bound = c
		} else {
			outerCap = c
		}
		return nil
	}
	running, since, phase, phaseAt := "", time.Time{}, "", time.Time{}
	var agains []string // the node of each run that ended on a transport fault, in order (ADR 0664 §3)
	standStill := func() error {
		if phase == "" {
			return fmt.Errorf("the ask %s's step on %s did not end within %s of its start: it stood still, saying no phase", a.ID, running, k.Bound)
		}
		return fmt.Errorf("the ask %s's step on %s did not end within %s: it stood still in %q since %s",
			a.ID, running, k.Bound, phase, phaseAt.UTC().Format(time.RFC3339))
	}
	for {
		select {
		case <-ctx.Done():
			var w Withdrawn
			if errors.As(context.Cause(ctx), &w) {
				k.dropped(a, context.Cause(ctx).Error())
			}
			return buildstep.Step{}, ctx.Err()
		case <-bound:
			if len(agains) > 0 && phase == "given again after a transport fault" {
				k.dropped(a, "the bus did not give it again within its bound")
			}
			return buildstep.Step{}, standStill()
		case <-outerCap:
			return buildstep.Step{}, fmt.Errorf("the ask %s's step on %s did not end within %s of its start: it kept moving (last phase %q at %s) but never ended",
				a.ID, running, 3*k.Bound, phase, phaseAt.UTC().Format(time.RFC3339))
		case step, ok := <-steps:
			if !ok {
				return buildstep.Step{}, errors.New("the watch of the ask's step ended")
			}
			if seen := stepSeenOf(ctx); seen != nil {
				seen(step)
			}
			if step.State == buildstep.StepAgain {
				// A run ended on a transport fault and the bus gives the ask again: not an end. The bound runs on from the
				// time it is given again, so a bus that never gives it says so (ADR 0664 §3).
				agains = append(agains, step.Node)
				running, since, phase, phaseAt = step.Node, step.Started, "given again after a transport fault", step.AgainAt
				if k.Bound > 0 {
					if err := arm(0, time.Until(step.AgainAt.Add(k.Bound))); err != nil {
						return buildstep.Step{}, fmt.Errorf("the bound of the ask %s's step cannot be kept: %w", a.ID, err)
					}
				}
				continue
			}
			if buildstep.Ended(step) {
				if step.Passing && step.State == buildstep.StepFailed && (len(agains) > 0 || step.Run > 1) {
					// The step's own count of runs (ADR 0664 §2), not only the ones this wait saw: a builder that started again
					// reads the key's last value and none of the runs before it.
					seat, _ := buildstep.BuildSeat(a.Language)
					step.Reason = TransportFault(FaultClass(step.Reason), seat, step.Node, step.Reason)
				}
				return step, nil
			}
			if k.Bound <= 0 {
				continue
			}
			newRun := step.Node != running || !step.Started.Equal(since)
			if !newRun && step.Phase == phase && step.PhaseAt.Equal(phaseAt) {
				continue
			}
			// A node took the ask, or took it again after its run stopped (another node, or the same node's build module
			// replaced mid-run: a new start), or its step entered a phase. The bound is on standing still: it runs from the
			// step's last movement (its phase's start, else its start), so a step that keeps moving is waited for, up to
			// three bounds from its start (issue 1930).
			running, since, phase, phaseAt = step.Node, step.Started, step.Phase, step.PhaseAt
			last := step.Started
			if step.PhaseAt.After(last) {
				last = step.PhaseAt
			}
			left := k.Bound
			if !last.IsZero() {
				left = time.Until(last.Add(k.Bound))
			}
			if err := arm(0, left); err != nil {
				return buildstep.Step{}, fmt.Errorf("the bound of the ask %s's step cannot be kept: %w", a.ID, err)
			}
			if newRun {
				outer := 3 * k.Bound
				if !step.Started.IsZero() {
					outer = time.Until(step.Started.Add(3 * k.Bound))
				}
				if err := arm(1, outer); err != nil {
					return buildstep.Step{}, fmt.Errorf("the bound of the ask %s's step cannot be kept: %w", a.ID, err)
				}
			}
		}
	}
}

type stepSeenKey struct{}

// WithStepSeen is ctx carrying seen, which Asker.Wait hands every value the ask's step key takes while it waits: the node
// that took it, each phase it says while running (buildstep.Step.Phase), and its end. A try names its asks' progress so
// (issue 1890). seen runs in the waiting goroutine and must not block. A seen ctx already carries is kept: both are handed
// every value, the earlier first (a check's begin bound watches the steps of a try that names them).
func WithStepSeen(ctx context.Context, seen func(buildstep.Step)) context.Context {
	if before := stepSeenOf(ctx); before != nil {
		later := seen
		seen = func(s buildstep.Step) { before(s); later(s) }
	}
	return context.WithValue(ctx, stepSeenKey{}, seen)
}

func stepSeenOf(ctx context.Context) func(buildstep.Step) {
	seen, _ := ctx.Value(stepSeenKey{}).(func(buildstep.Step))
	return seen
}

// Ask puts a and waits for its step; a step that did not pass is an error naming its node, its state and its reason.
func (k Asker) Ask(ctx context.Context, a buildstep.StepAsk) (buildstep.Step, error) {
	if err := k.Put(ctx, a); err != nil {
		return buildstep.Step{}, err
	}
	return k.Passed(k.Wait(ctx, a))
}

// Passed is step when it passed, else an error naming where it ended and why.
func (k Asker) Passed(step buildstep.Step, err error) (buildstep.Step, error) {
	if err != nil {
		return step, err
	}
	if step.State != buildstep.StepPassed {
		return step, fmt.Errorf("the ask %s on %s %s: %s", step.ID, step.Node, step.State, step.Reason)
	}
	return step, nil
}

// JetStreamAsks is the bus side of an Asker: a JetStream publish de-duplicated by the ask's id (Nats-Msg-Id), and a watch
// of `*.<ask id>` in the build module's steps bucket.
type JetStreamAsks struct {
	JS jetstream.JetStream
	// Unanswered is told each time the read of an ask's steps meets a transport fault it keeps waiting out (design 47 G §568, ADR
	// 0668 §3): the builder arms the call's bound, a key with its own time to live (A-77), and raises call-unanswered when it
	// ends unanswered. Answered is told when the read is answered again. Both may be nil.
	Unanswered func(module, askID string, fault error)
	Answered   func(module, askID string)
	// OpenWatch opens the watch of keys filtered by pattern in the build module's steps state, from the revision after `after`
	// when after is not 0; nil is the bus's.
	OpenWatch func(ctx context.Context, module, pattern string, after uint64) (jetstream.KeyWatcher, error)
}

// PublishAsk is AskPublisher's.
func (s JetStreamAsks) PublishAsk(ctx context.Context, subject, id string, data []byte) error {
	msg := &nats.Msg{Subject: subject, Data: data, Header: nats.Header{}}
	msg.Header.Set(jetstream.MsgIDHeader, id)
	_, err := s.JS.PublishMsg(ctx, msg)
	return err
}

// openWatch opens the watch of pattern in the steps state of module, from the revision after `after` when it is not 0.
func (s JetStreamAsks) openWatch(ctx context.Context, module, pattern string, after uint64) (jetstream.KeyWatcher, error) {
	if s.OpenWatch != nil {
		return s.OpenWatch(ctx, module, pattern, after)
	}
	kv, err := s.JS.KeyValue(ctx, StepsBucket(module))
	if err != nil {
		return nil, fmt.Errorf("the bucket %s: %w", StepsBucket(module), err)
	}
	if after > 0 {
		return kv.Watch(ctx, pattern, jetstream.ResumeFromRevision(after))
	}
	return kv.Watch(ctx, pattern)
}

// transportFault says err is a transport fault of ADR 0664 §1's classes or the read's own deadline: nothing was answered.
func transportFault(err error) bool {
	return busretry.Undelivered(err) || busretry.TimedOut(err)
}

// openKept is a kept call: the watch is opened, and opened again while the bus answers nothing, until it is or ctx ends
// (design 47 G §568: a transport fault of the builder's own read ends no build key). Each fault past the call's own asking again
// is told to Unanswered; the answer, to Answered. An error that is no transport fault is the caller's.
func (s JetStreamAsks) openKept(ctx context.Context, module, askID, pattern string, after uint64) (jetstream.KeyWatcher, error) {
	for {
		w, err := busretry.DoRepeatable("opening the watch of the ask "+askID+"'s step on "+module, func() (jetstream.KeyWatcher, error) {
			return s.openWatch(ctx, module, pattern, after)
		}, nil)
		if err == nil {
			if s.Answered != nil {
				s.Answered(module, askID)
			}
			return w, nil
		}
		if ctx.Err() != nil {
			return nil, ctx.Err()
		}
		if !transportFault(err) {
			return nil, err
		}
		if s.Unanswered != nil {
			s.Unanswered(module, askID, err)
		}
	}
}

// WatchAsk is AskWaiter's: the watch hands over what the keys hold now, so a step that ended before it began is read at
// once, and every change after. A watch that cannot be opened for a transport fault is opened again once the bus answers, and one
// that closes is taken up again from the revision after the last it handed over (design 47 G §568), so a fault fails no key.
func (s JetStreamAsks) WatchAsk(ctx context.Context, module, askID string) (<-chan buildstep.Step, error) {
	pattern, err := buildstep.AskStepWatch(askID)
	if err != nil {
		return nil, err
	}
	w, err := s.openKept(ctx, module, askID, pattern, 0)
	if err != nil {
		return nil, err
	}
	out := make(chan buildstep.Step)
	go func() {
		defer close(out)
		var last uint64
		for {
			closed := false
			for !closed {
				select {
				case <-ctx.Done():
					_ = w.Stop()
					return
				case e, ok := <-w.Updates():
					if !ok {
						closed = true
						break
					}
					if e == nil || e.Operation() != jetstream.KeyValuePut {
						continue
					}
					last = max(last, e.Revision())
					step, err := buildstep.DecodeStep(e.Value())
					if err != nil {
						continue // not a step (a node's pause key never matches an ask id; anything else is not this ask's)
					}
					select {
					case out <- step:
					case <-ctx.Done():
						_ = w.Stop()
						return
					}
				}
			}
			_ = w.Stop()
			if w, err = s.openKept(ctx, module, askID, pattern, last+1); err != nil {
				return
			}
		}
	}()
	return out, nil
}

// FakeAsks is a test double of the bus for an asker's tests: each ask put is kept and, when Answer is set, its step is
// written at once with what Answer says a build node wrote (running or ended). Write writes a step later. Safe for
// concurrent use.
type FakeAsks struct {
	Answer func(a buildstep.StepAsk) buildstep.Step

	mu       sync.Mutex
	asks     []buildstep.StepAsk
	steps    map[string][]buildstep.Step
	watchers map[string][]chan buildstep.Step
}

// PublishAsk is AskPublisher's.
func (f *FakeAsks) PublishAsk(_ context.Context, _, id string, data []byte) error {
	a, err := buildstep.DecodeStepAsk(data)
	if err != nil {
		return err
	}
	f.mu.Lock()
	f.asks = append(f.asks, a)
	f.mu.Unlock()
	if f.Answer != nil {
		f.Write(f.Answer(a))
	}
	return nil
}

// Write writes an ask's step, as a build node does.
func (f *FakeAsks) Write(step buildstep.Step) {
	f.mu.Lock()
	if f.steps == nil {
		f.steps, f.watchers = map[string][]buildstep.Step{}, map[string][]chan buildstep.Step{}
	}
	f.steps[step.ID] = append(f.steps[step.ID], step)
	ws := append([]chan buildstep.Step(nil), f.watchers[step.ID]...)
	f.mu.Unlock()
	for _, w := range ws {
		w <- step
	}
}

// End is Write: kept for the tests that end a step.
func (f *FakeAsks) End(step buildstep.Step) { f.Write(step) }

// WatchAsk is AskWaiter's: what was written so far, then every write.
func (f *FakeAsks) WatchAsk(ctx context.Context, _ string, askID string) (<-chan buildstep.Step, error) {
	out := make(chan buildstep.Step, 64)
	f.mu.Lock()
	if f.steps == nil {
		f.steps, f.watchers = map[string][]buildstep.Step{}, map[string][]chan buildstep.Step{}
	}
	for _, s := range f.steps[askID] {
		out <- s
	}
	f.watchers[askID] = append(f.watchers[askID], out)
	f.mu.Unlock()
	go func() {
		<-ctx.Done()
		f.mu.Lock()
		ws := f.watchers[askID]
		for i, w := range ws {
			if w == out {
				f.watchers[askID] = append(ws[:i], ws[i+1:]...)
				break
			}
		}
		f.mu.Unlock()
	}()
	return out, nil
}

// Asks are the asks put so far.
func (f *FakeAsks) Asks() []buildstep.StepAsk {
	f.mu.Lock()
	defer f.mu.Unlock()
	return append([]buildstep.StepAsk(nil), f.asks...)
}