221 lines
6.5 KiB
Go
221 lines
6.5 KiB
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 waits for an ask's step: the first key `*.<ask id>` of the build module's steps state whose state is not
|
|
// running. A watch, not a poll; it returns when the step ends or ctx does.
|
|
type AskWaiter interface {
|
|
WaitAsk(ctx context.Context, module, askID string) (delivery.Step, error)
|
|
}
|
|
|
|
// Asker puts asks and waits for their steps.
|
|
type Asker struct {
|
|
Publisher AskPublisher
|
|
Waiter AskWaiter
|
|
// Bound is how long the asker waits for an ask's step before it stops waiting; zero waits as long as ctx.
|
|
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.
|
|
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
|
|
}
|
|
wait, cancel := ctx, context.CancelFunc(func() {})
|
|
if k.Bound > 0 {
|
|
wait, cancel = context.WithTimeoutCause(ctx, k.Bound, fmt.Errorf("the ask %s's step did not end within %s", a.ID, k.Bound))
|
|
}
|
|
defer cancel()
|
|
step, err := k.Waiter.WaitAsk(wait, module, a.ID)
|
|
if err != nil {
|
|
if cause := context.Cause(wait); cause != nil && ctx.Err() == nil {
|
|
err = cause
|
|
}
|
|
return delivery.Step{}, err
|
|
}
|
|
return step, nil
|
|
}
|
|
|
|
// 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
|
|
}
|
|
|
|
// WaitAsk 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) WaitAsk(ctx context.Context, module, askID string) (delivery.Step, error) {
|
|
pattern, err := delivery.AskStepWatch(askID)
|
|
if err != nil {
|
|
return delivery.Step{}, err
|
|
}
|
|
kv, err := s.JS.KeyValue(ctx, delivery.StepsBucket(module))
|
|
if err != nil {
|
|
return delivery.Step{}, fmt.Errorf("the bucket %s: %w", delivery.StepsBucket(module), err)
|
|
}
|
|
w, err := kv.Watch(ctx, pattern)
|
|
if err != nil {
|
|
return delivery.Step{}, err
|
|
}
|
|
defer w.Stop()
|
|
for {
|
|
select {
|
|
case <-ctx.Done():
|
|
return delivery.Step{}, ctx.Err()
|
|
case e, ok := <-w.Updates():
|
|
if !ok {
|
|
return delivery.Step{}, errors.New("the watch of the ask's step ended")
|
|
}
|
|
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)
|
|
}
|
|
if step.State != delivery.StepRunning {
|
|
return step, 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
|
|
// ended at once with what Answer says a build node wrote. 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
|
|
waits map[string]chan struct{}
|
|
}
|
|
|
|
// 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.End(f.Answer(a))
|
|
}
|
|
return nil
|
|
}
|
|
|
|
// End writes an ask's step end, as a build node does.
|
|
func (f *FakeAsks) End(step delivery.Step) {
|
|
f.mu.Lock()
|
|
if f.steps == nil {
|
|
f.steps, f.waits = map[string]delivery.Step{}, map[string]chan struct{}{}
|
|
}
|
|
f.steps[step.ID] = step
|
|
ch := f.waits[step.ID]
|
|
delete(f.waits, step.ID)
|
|
f.mu.Unlock()
|
|
if ch != nil {
|
|
close(ch)
|
|
}
|
|
}
|
|
|
|
// WaitAsk is AskWaiter's.
|
|
func (f *FakeAsks) WaitAsk(ctx context.Context, _ string, askID string) (delivery.Step, error) {
|
|
f.mu.Lock()
|
|
if f.steps == nil {
|
|
f.steps, f.waits = map[string]delivery.Step{}, map[string]chan struct{}{}
|
|
}
|
|
if s, ok := f.steps[askID]; ok {
|
|
f.mu.Unlock()
|
|
return s, nil
|
|
}
|
|
ch := f.waits[askID]
|
|
if ch == nil {
|
|
ch = make(chan struct{})
|
|
f.waits[askID] = ch
|
|
}
|
|
f.mu.Unlock()
|
|
select {
|
|
case <-ctx.Done():
|
|
return delivery.Step{}, ctx.Err()
|
|
case <-ch:
|
|
}
|
|
f.mu.Lock()
|
|
defer f.mu.Unlock()
|
|
return f.steps[askID], 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...)
|
|
}
|