mesh-sdk/go/delivery/buildrun/asks.go

package buildrun

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

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

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

// **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 delivery.Step, error)
}

// Asker puts asks and waits for their steps.
type Asker struct {
	Publisher AskPublisher
	Waiter    AskWaiter
	// Bound is how long a step may run once a build node took the ask (its key written running) before the asker stops
	// waiting; 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
}

// Put puts a on its stream: the seat's accept subject of its language and kind.
func (k Asker) Put(ctx context.Context, a delivery.StepAsk) error {
	raw, err := delivery.EncodeStepAsk(a)
	if err != nil {
		return err
	}
	subject, err := delivery.AskSubject(a.Language, a.Kind)
	if err != nil {
		return err
	}
	if err := k.Publisher.PublishAsk(ctx, subject, a.ID, raw); err != nil {
		return fmt.Errorf("putting the %s ask %s: %w", a.Kind, a.ID, err)
	}
	return nil
}

// 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 delivery.StepAsk) (delivery.Step, error) {
	module, err := delivery.BuildModuleOf(a.Language)
	if err != nil {
		return delivery.Step{}, err
	}
	watch, cancel := context.WithCancel(ctx)
	defer cancel()
	steps, err := k.Waiter.WatchAsk(watch, module, a.ID)
	if err != nil {
		return delivery.Step{}, err
	}
	var bound <-chan time.Time
	var timer *time.Timer
	defer func() {
		if timer != nil {
			timer.Stop()
		}
	}()
	running, since := "", time.Time{}
	for {
		select {
		case <-ctx.Done():
			return delivery.Step{}, ctx.Err()
		case <-bound:
			return delivery.Step{}, fmt.Errorf("the ask %s's step on %s did not end within %s of its start", a.ID, running, k.Bound)
		case step, ok := <-steps:
			if !ok {
				return delivery.Step{}, errors.New("the watch of the ask's step ended")
			}
			if step.State != delivery.StepRunning {
				return step, nil
			}
			if k.Bound <= 0 || (step.Node == running && step.Started.Equal(since)) {
				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): its step is bounded from its own start.
			running, since = step.Node, step.Started
			left := k.Bound
			if !step.Started.IsZero() {
				left = time.Until(step.Started.Add(k.Bound))
			}
			if timer != nil {
				timer.Stop()
			}
			timer = time.NewTimer(max(left, 0))
			bound = timer.C
		}
	}
}

// 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 delivery.StepAsk) (delivery.Step, error) {
	if err := k.Put(ctx, a); err != nil {
		return delivery.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 delivery.Step, err error) (delivery.Step, error) {
	if err != nil {
		return step, err
	}
	if step.State != delivery.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
}

// 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
}

// 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.
func (s JetStreamAsks) WatchAsk(ctx context.Context, module, askID string) (<-chan delivery.Step, error) {
	pattern, err := delivery.AskStepWatch(askID)
	if err != nil {
		return nil, err
	}
	kv, err := s.JS.KeyValue(ctx, delivery.StepsBucket(module))
	if err != nil {
		return nil, fmt.Errorf("the bucket %s: %w", delivery.StepsBucket(module), err)
	}
	w, err := kv.Watch(ctx, pattern)
	if err != nil {
		return nil, err
	}
	out := make(chan delivery.Step)
	go func() {
		defer close(out)
		defer w.Stop()
		for {
			select {
			case <-ctx.Done():
				return
			case e, ok := <-w.Updates():
				if !ok {
					return
				}
				if e == nil || e.Operation() != jetstream.KeyValuePut {
					continue
				}
				step, err := delivery.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():
					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 delivery.StepAsk) delivery.Step

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

// PublishAsk is AskPublisher's.
func (f *FakeAsks) PublishAsk(_ context.Context, _, id string, data []byte) error {
	a, err := delivery.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 delivery.Step) {
	f.mu.Lock()
	if f.steps == nil {
		f.steps, f.watchers = map[string][]delivery.Step{}, map[string][]chan delivery.Step{}
	}
	f.steps[step.ID] = append(f.steps[step.ID], step)
	ws := append([]chan delivery.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 delivery.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 delivery.Step, error) {
	out := make(chan delivery.Step, 64)
	f.mu.Lock()
	if f.steps == nil {
		f.steps, f.watchers = map[string][]delivery.Step{}, map[string][]chan delivery.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() []delivery.StepAsk {
	f.mu.Lock()
	defer f.mu.Unlock()
	return append([]delivery.StepAsk(nil), f.asks...)
}