jschoubben
ffc4b88859
The indexer's validation ask rides accept.<queue>.<asker> under its own name (§712, A-232); an asker that names none still puts under the builder's. Builds: design 47 G A-248, A-247, §712 Claude-Session: https://claude.ai/code/session_01NcbgfGKaxw984XyAPfLm4h
609 lines
23 KiB
Go
609 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)
|
|
}
|
|
|
|
// 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)
|
|
}
|
|
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...)
|
|
}
|