jschoubben
93ecfc7f1f
The shared asker (buildrun.Asker.Wait) answers an ended step whole: a key naming its answer as output-object has it read through the build seat's verb output on the step's node (Step.Whole) with the asker's Outputs, so the checker's judge, manifests judge, checking program and compose sandbox read the answer as before; a step whose object cannot be read is the wait's error. The checker reads its run asks' failed key's answer for a program that exited other than 0. The checker, the indexer and the assigner read a kept answer through process.Conn.CallFile (A-291), which receives the verb's chunks; the checker's manifest invokes the three build seats' output. Builds: design 47 G A-290, A-291, A-254 Claude-Session: https://claude.ai/code/session_01NcbgfGKaxw984XyAPfLm4h
617 lines
23 KiB
Go
617 lines
23 KiB
Go
package buildrun
|
|
|
|
import (
|
|
"cmp"
|
|
"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/modules/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 {
|
|
// Name is the asker the subject names, the caller's seat name (design 47 G A-248, A-163): empty puts under the builder's
|
|
// name until the asker rename after the join.
|
|
Name string
|
|
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)
|
|
// Outputs reads the answer a step's key names as its output-object, being over what the steps state holds (design 47 G
|
|
// A-290), through the build seat's verb output on the step's node (buildstep.Step.Whole); Wait answers the step whole.
|
|
// Nil reads none: such a step's wait is an error saying so.
|
|
Outputs buildstep.AskVerb
|
|
}
|
|
|
|
// 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 asker's Name, else the builder's
|
|
// name (design 47 G §481, §600, A-248: `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, cmp.Or(k.Name, 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"
|
|
}
|
|
if d < time.Second { // a key's time to live is at least a second; a bound already past ends at once
|
|
if d <= time.Millisecond {
|
|
k.Expiry.Ended(key)
|
|
return nil
|
|
}
|
|
d = time.Second
|
|
}
|
|
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)
|
|
}
|
|
whole, err := step.Whole(a.Language, a.Run != nil, k.Outputs)
|
|
if err != nil {
|
|
return buildstep.Step{}, fmt.Errorf("the ask %s's step on %s ended %s, and its answer %s cannot be read: %w", a.ID, step.Node, step.State, step.OutputObject, err)
|
|
}
|
|
return whole, 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...)
|
|
}
|