From 88fa74ce5e3c91b62464ee3dd9fca3a8ab0842bd Mon Sep 17 00:00:00 2001 From: jochen Date: Fri, 23 Oct 2026 16:32:21 +0200 Subject: [PATCH] =?UTF-8?q?Bound=20an=20ask's=20step=20from=20its=20start?= =?UTF-8?q?=20on=20the=20node=20that=20took=20it,=20not=20from=20its=20put?= =?UTF-8?q?,=20so=20an=20ask=20waiting=20in=20its=20stream=20during=20a=20?= =?UTF-8?q?large=20rebuild=20is=20never=20failed=20and=20asked=20twice=20(?= =?UTF-8?q?issue=201887,=20review=20round=202=20item=201;=20ADR=200525=20?= =?UTF-8?q?=C2=A73,=20=C2=A76)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- go/delivery/buildrun/asks.go | 186 ++++++++++++++++++------------ go/delivery/buildrun/asks_test.go | 12 +- 2 files changed, 125 insertions(+), 73 deletions(-) diff --git a/go/delivery/buildrun/asks.go b/go/delivery/buildrun/asks.go index de6dac2..97607fe 100644 --- a/go/delivery/buildrun/asks.go +++ b/go/delivery/buildrun/asks.go @@ -22,17 +22,19 @@ type AskPublisher interface { PublishAsk(ctx context.Context, subject, id string, data []byte) error } -// AskWaiter waits for an ask's step: the first key `*.` 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. +// AskWaiter watches an ask's steps: every value a key `*.` 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 { - WaitAsk(ctx context.Context, module, askID string) (delivery.Step, error) + 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 the asker waits for an ask's step before it stops waiting; zero waits as long as ctx. + // 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 } @@ -53,25 +55,56 @@ func (k Asker) Put(ctx context.Context, a delivery.StepAsk) error { } // 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. +// 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 } - 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)) - } + watch, cancel := context.WithCancel(ctx) defer cancel() - step, err := k.Waiter.WaitAsk(wait, module, a.ID) + steps, err := k.Waiter.WatchAsk(watch, 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 + var bound <-chan time.Time + var timer *time.Timer + defer func() { + if timer != nil { + timer.Stop() + } + }() + running := "" + 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 { + continue + } + // A node took the ask (or another took it after the first stopped): its step is bounded from its start. + running = step.Node + 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. @@ -107,53 +140,61 @@ func (s JetStreamAsks) PublishAsk(ctx context.Context, subject, id string, data 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 +// 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) WaitAsk(ctx context.Context, module, askID string) (delivery.Step, error) { +func (s JetStreamAsks) WatchAsk(ctx context.Context, module, askID string) (<-chan delivery.Step, error) { pattern, err := delivery.AskStepWatch(askID) if err != nil { - return delivery.Step{}, err + return nil, 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) + return nil, fmt.Errorf("the bucket %s: %w", delivery.StepsBucket(module), err) } w, err := kv.Watch(ctx, pattern) if err != nil { - return delivery.Step{}, err + return nil, 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 + 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 -// ended at once with what Answer says a build node wrote. Safe for concurrent use. +// 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 - waits map[string]chan struct{} + mu sync.Mutex + asks []delivery.StepAsk + steps map[string][]delivery.Step + watchers map[string][]chan delivery.Step } // PublishAsk is AskPublisher's. @@ -166,50 +207,53 @@ func (f *FakeAsks) PublishAsk(_ context.Context, _, id string, data []byte) erro f.asks = append(f.asks, a) f.mu.Unlock() if f.Answer != nil { - f.End(f.Answer(a)) + f.Write(f.Answer(a)) } return nil } -// End writes an ask's step end, as a build node does. -func (f *FakeAsks) End(step delivery.Step) { +// 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.waits = map[string]delivery.Step{}, map[string]chan struct{}{} + f.steps, f.watchers = map[string][]delivery.Step{}, map[string][]chan delivery.Step{} } - f.steps[step.ID] = step - ch := f.waits[step.ID] - delete(f.waits, step.ID) + f.steps[step.ID] = append(f.steps[step.ID], step) + ws := append([]chan delivery.Step(nil), f.watchers[step.ID]...) f.mu.Unlock() - if ch != nil { - close(ch) + for _, w := range ws { + w <- step } } -// WaitAsk is AskWaiter's. -func (f *FakeAsks) WaitAsk(ctx context.Context, _ string, askID string) (delivery.Step, error) { +// 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.waits = map[string]delivery.Step{}, map[string]chan struct{}{} + f.steps, f.watchers = map[string][]delivery.Step{}, map[string][]chan delivery.Step{} } - 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 + for _, s := range f.steps[askID] { + out <- s } + f.watchers[askID] = append(f.watchers[askID], out) 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 + 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. diff --git a/go/delivery/buildrun/asks_test.go b/go/delivery/buildrun/asks_test.go index b34466f..9432242 100644 --- a/go/delivery/buildrun/asks_test.go +++ b/go/delivery/buildrun/asks_test.go @@ -32,10 +32,18 @@ func TestAnAskerPutsOnceAndReadsTheStep(t *testing.T) { if _, err := k.Ask(context.Background(), goAsk("mvl0frv18d1958")); err == nil || !strings.Contains(err.Error(), "shanks refused: go 1.27 installed") { t.Fatalf("a refused step: %v", err) } + // review round 2 item 1: an ask that waits in its stream for a free slot is not bounded; once a node writes its step + // running, the step is bounded from its start, and the error names the node. g := &FakeAsks{} k = Asker{Publisher: g, Waiter: g, Bound: 20 * time.Millisecond} - if _, err := k.Wait(context.Background(), goAsk("mvl0frv18d1959")); err == nil || !strings.Contains(err.Error(), "did not end within") { - t.Fatalf("an ask whose step never ends: %v", err) + waiting, cancel := context.WithTimeout(context.Background(), 100*time.Millisecond) + defer cancel() + if _, err := k.Wait(waiting, goAsk("mvl0frv18d1959")); err != context.DeadlineExceeded { + t.Fatalf("an ask no node took: %v, want it still waiting when ctx ended", err) + } + g.Write(delivery.Step{ID: "mvl0frv18d1961", Node: "ace", State: delivery.StepRunning, Started: time.Now()}) + if _, err := k.Wait(context.Background(), goAsk("mvl0frv18d1961")); err == nil || !strings.Contains(err.Error(), "on ace did not end within") { + t.Fatalf("a step that never ends: %v", err) } if len(g.Asks()) != 0 { t.Fatal("a wait put an ask")