Bound an ask's step from its start on the node that took it, not from its put, so an ask waiting in its stream during a large rebuild is never failed and asked twice (issue 1887, review round 2 item 1; ADR 0525 §3, §6)

2026-10-23 16:32:21 +02:00
parent 9205222a98
commit 88fa74ce5e
2 changed files with 125 additions and 73 deletions
@@ -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 `*.<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.
+// 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 {
-	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.
@@ -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")