buildrun: the asker hears DEAD_LETTERS again on its next reconnect or ask, in the background, and says whether it hears; buscompose grants only what a pulled ordered consumer needs (issue 734, review round 1)

An asker started before its grant was composed was refused once and stayed deaf until it started again, silently. DeadLetterHearing tries in the background and again on an event (Again: a reconnect, an ask waited for while deaf), and tells each change of Hearing so the module's health tool can say it. DeadLetterReads drops the flow control a pull consumer does not use and deletes one consumer by name.
2026-10-26 14:16:58 +01:00
parent d96b9fd660
commit 9b01c94a9e
4 changed files with 249 additions and 40 deletions
@@ -85,6 +85,7 @@ func TestTheCopiedComposerComposesAsTheControllers(t *testing.T) {
 			"$JS.API.CONSUMER.CREATE.DEAD_LETTERS.x1.mesh.events.dead.EVENTS.>":                  false,
 			"$JS.API.CONSUMER.CREATE.DEAD_LETTERS.x1":                                            false,
 			"$JS.API.STREAM.PURGE.DEAD_LETTERS":                                                  false,
+			"$JS.FC.DEAD_LETTERS.x1":                                                             false,
 		} {
 			if got := MayPublish(perms, subject); got != want {
 				t.Errorf("may publish %s: %v, want %v", subject, got, want)
@@ -157,7 +157,7 @@ func SeatTrafficOf(module string, holds, uses, watches []Seat) SeatTraffic {

 // DeadLetterReads is what a user of a seat pulled by kind is granted to hear which of the asks it put the seat's workers
 // gave up on (novox/hq issue 734): an ordered consumer on DEAD_LETTERS filtered to the copies the controller keeps of that
-// seat's stream (DeadLetterSubject(stream, *)), pulled or pushed, its deletion and its flow control. A copy carries the ask's own message id
+// seat's stream (DeadLetterSubject(stream, *)), pulled, and its deletion. A copy carries the ask's own message id
 // (Mesh-Dead-Msg-Id), so the asker ends the step it waits for instead of waiting for ever; it reads the seat's dead
 // letters and no other stream's, and changes none.
 func DeadLetterReads(stream string) []string {
@@ -165,8 +165,7 @@ func DeadLetterReads(stream string) []string {
 		"$JS.API.STREAM.INFO." + DeadLettersStream,
 		"$JS.API.CONSUMER.CREATE." + DeadLettersStream + ".*." + DeadLetterSubject(stream, ">"),
 		"$JS.API.CONSUMER.MSG.NEXT." + DeadLettersStream + ".*",
-		"$JS.API.CONSUMER.DELETE." + DeadLettersStream + ".>",
-		"$JS.FC." + DeadLettersStream + ".>",
+		"$JS.API.CONSUMER.DELETE." + DeadLettersStream + ".*",
 	}
 }

@@ -2,7 +2,9 @@ package buildrun

 import (
 	"context"
+	"errors"
 	"strings"
+	"sync"
 	"testing"
 	"time"

@@ -77,6 +79,33 @@ func TestAnAskerPutsOnceAndReadsTheStep(t *testing.T) {
 	if step, err = k.Wait(NotTakenBound(context.Background(), time.Minute), goAsk("mvl0frv18d1972")); err != nil || step.Node != "g14" {
 		t.Fatalf("a step written: %+v, %v", step, err)
 	}
+	// Review round 1 of issue 734: a hearing the bus refuses at start is deaf and says so, never holds the start up, and
+	// tries again on the next ask waited for; heard, it says so and hands the given-up asks on.
+	fol := &refusingFollower{refuse: 1}
+	var mu sync.Mutex
+	var told []Hearing
+	hearing := NewDeadLetterHearing(context.Background(), fol, []string{delivery.AskGo}, dead, nil,
+		func(h Hearing) { mu.Lock(); told = append(told, h); mu.Unlock() })
+	hearing.Wait()
+	if why := hearing.Deaf(); !strings.Contains(why, "SEAT_NODE_BUILD_GO: permissions violation") {
+		t.Fatalf("a refused hearing says %q", why)
+	}
+	k = Asker{Publisher: g, Waiter: LostAskWaiter{Inner: g, Dead: dead, Hearing: hearing}, Bound: time.Minute}
+	g.Write(delivery.Step{ID: "mvl0frv18d1973", Node: "g14", State: delivery.StepPassed})
+	if _, err := k.Wait(context.Background(), goAsk("mvl0frv18d1973")); err != nil {
+		t.Fatal(err)
+	}
+	hearing.Wait()
+	mu.Lock()
+	if hearing.Deaf() != "" || len(told) != 2 || told[0].Heard || !told[1].Heard || fol.calls() != 2 {
+		t.Fatalf("after an ask: deaf %q, told %+v, followed %d times", hearing.Deaf(), told, fol.calls())
+	}
+	mu.Unlock()
+	fol.heard(GivenUp{AskID: "mvl0frv18d1974", Deliveries: 5})
+	step, err = k.Wait(context.Background(), goAsk("mvl0frv18d1974"))
+	if err != nil || !IsLost(step) {
+		t.Fatalf("an ask heard given up after the hearing came back: %+v, %v", step, err)
+	}
 }

 // Review (issue 734 path): an ask taken again by the same node (its build module replaced mid-run) is bounded from the
@@ -98,3 +127,26 @@ func TestARedeliveryToTheSameNodeRestartsTheBound(t *testing.T) {
 		t.Fatalf("step %+v, %v, want the redelivered run's pass", step, err)
 	}
 }
+
+// refusingFollower refuses the first `refuse` follows, as the bus does before the asker's grant is composed.
+type refusingFollower struct {
+	mu     sync.Mutex
+	refuse int
+	n      int
+	hand   func(GivenUp)
+}
+
+func (f *refusingFollower) Follow(_ context.Context, _ string, _ time.Time, heard func(GivenUp)) (func(), error) {
+	f.mu.Lock()
+	defer f.mu.Unlock()
+	f.n++
+	if f.n <= f.refuse {
+		return nil, errors.New("permissions violation")
+	}
+	f.hand = heard
+	return func() {}, nil
+}
+
+func (f *refusingFollower) calls() int { f.mu.Lock(); defer f.mu.Unlock(); return f.n }
+
+func (f *refusingFollower) heard(g GivenUp) { f.mu.Lock(); h := f.hand; f.mu.Unlock(); h(g) }
@@ -20,7 +20,7 @@ import (
 // was given up. No build node writes a step for it, so an asker that waits for the step waits for ever: a try stays
 // running and holds a try place, a check's verdict stays running and its pull request waits.
 //
-// An asker of the build seats (the checker, the builder) hears those copies (HearDeadLetters: an ordered consumer on
+// An asker of the build seats (the checker, the builder) hears those copies (DeadLetterHearing: an ordered consumer on
 // DEAD_LETTERS filtered to the build seats' streams, granted to a user of a seat pulled by kind, buscompose
 // DeadLetterReads) into DeadLetters, and LostAskWaiter ends the waiting step at once as failed on the node LostNode, its
 // reason saying LostWords; the asker decides what follows (the checker ends a try as error and asks a check again, the
@@ -172,7 +172,13 @@ func notTakenBoundOf(ctx context.Context) time.Duration {
 type LostAskWaiter struct {
 	Inner AskWaiter
 	Dead  *DeadLetters
-	Now   func() time.Time
+	// Hearing, when set, is asked to hear DEAD_LETTERS again each time an ask is waited for while it cannot: an ask put is
+	// the event that makes hearing matter.
+	Hearing interface {
+		Deaf() string
+		Again()
+	}
+	Now func() time.Time
 }

 func (w LostAskWaiter) now() time.Time {
@@ -184,6 +190,9 @@ func (w LostAskWaiter) now() time.Time {

 // WatchAsk is AskWaiter's.
 func (w LostAskWaiter) WatchAsk(ctx context.Context, module, askID string) (<-chan delivery.Step, error) {
+	if w.Hearing != nil && w.Hearing.Deaf() != "" {
+		w.Hearing.Again()
+	}
 	inner, err := w.Inner.WatchAsk(ctx, module, askID)
 	if err != nil {
 		return nil, err
@@ -255,42 +264,190 @@ func IsNotTaken(s delivery.Step) bool {
 	return s.Node == LostNode && strings.Contains(s.Reason, NotTakenWords)
 }

-// HearDeadLetters follows, for each language's build seat stream, the copies DEAD_LETTERS keeps of the asks its workers
-// gave up on, into dead: an ordered consumer from now, which resumes where it was after a reconnect, and starts here again
-// when it is reset before it heard any, so a bus reload loses none. Where the bus refuses it (the grant not yet composed)
-// it says so and goes on; the stop it answers ends every consumer it started.
-func HearDeadLetters(ctx context.Context, js jetstream.JetStream, languages []string, dead *DeadLetters, logf func(string, ...any)) func() {
-	var stops []func()
-	since := time.Now()
-	for _, language := range languages {
-		stream := SeatStream(language)
-		filter := DeadLetterFilter(stream)
-		notHeard := func(err error) {
-			logf("the asks %s gives up on are not heard (%s of %s), so a wait for one ends only at its bound: %v",
-				stream, filter, DeadLettersStream, err)
-		}
-		cons, err := js.OrderedConsumer(ctx, DeadLettersStream, jetstream.OrderedConsumerConfig{
-			FilterSubjects: []string{filter}, DeliverPolicy: jetstream.DeliverByStartTimePolicy, OptStartTime: &since})
-		if err != nil {
-			notHeard(err)
-			continue
-		}
-		cc, err := cons.Consume(func(m jetstream.Msg) {
-			if g, ok := GivenUpOf(m.Headers()); ok {
-				logf("the ask %s on %s was given up after %d deliveries at %s: the wait for its step ends", g.AskID, g.Stream,
-					g.Deliveries, g.At.Format(time.RFC3339))
-				dead.Heard(g)
-			}
-		}, jetstream.ConsumeErrHandler(func(_ jetstream.ConsumeContext, err error) { logf("hearing %s: %v", DeadLettersStream, err) }))
-		if err != nil {
-			notHeard(err)
-			continue
-		}
-		stops = append(stops, cc.Stop)
+// **The asker hears DEAD_LETTERS, tries again when it cannot, and says whether it hears** (issue 734). The read is
+// granted with the asker's seat traffic (buscompose DeadLetterReads); an asker started before its grant was composed is
+// refused, and an asker that only tried once would stay deaf until it starts again, silently. DeadLetterHearing tries in
+// the background, never holding the start up, and again on an event (Again): the asker's reconnect to the bus, and every
+// ask waited for while it is deaf (LostAskWaiter.Hearing); no timer. Each change of whether it hears is said (Said), so
+// the module's health tool says it: a deaf asker is not healthy.
+
+// Hearing is what a DeadLetterHearing says: whether it hears every build seat stream it follows, and why not.
+type Hearing struct {
+	Heard   bool      `json:"heard"`
+	Streams []string  `json:"streams"`
+	Why     string    `json:"why,omitempty"`
+	At      time.Time `json:"at"`
+}
+
+// HearingKey is the key a module keeps its Hearing under, in its own state (the checker's and the builder's hearing).
+const HearingKey = "dead-letters"
+
+// DeadLetterFollower starts following one seat stream's copies in DEAD_LETTERS from since, handing each given-up ask to
+// heard; it answers the stop, or why it cannot.
+type DeadLetterFollower interface {
+	Follow(ctx context.Context, stream string, since time.Time, heard func(GivenUp)) (func(), error)
+}
+
+// JetStreamDeadLetters is the bus side of a DeadLetterFollower: an ordered consumer on DEAD_LETTERS filtered to the
+// stream's copies, which resumes where it was after a reconnect and, reset before it heard any, starts at since again.
+type JetStreamDeadLetters struct{ JS jetstream.JetStream }
+
+// Follow is DeadLetterFollower's.
+func (j JetStreamDeadLetters) Follow(ctx context.Context, stream string, since time.Time, heard func(GivenUp)) (func(), error) {
+	cons, err := j.JS.OrderedConsumer(ctx, DeadLettersStream, jetstream.OrderedConsumerConfig{
+		FilterSubjects: []string{DeadLetterFilter(stream)}, DeliverPolicy: jetstream.DeliverByStartTimePolicy, OptStartTime: &since})
+	if err != nil {
+		return nil, err
 	}
-	return func() {
-		for _, stop := range stops {
-			stop()
+	cc, err := cons.Consume(func(m jetstream.Msg) {
+		if g, ok := GivenUpOf(m.Headers()); ok {
+			heard(g)
 		}
+	})
+	if err != nil {
+		return nil, err
+	}
+	return cc.Stop, nil
+}
+
+// DeadLetterHearing follows the build seats' given-up asks into Dead. Make it with NewDeadLetterHearing.
+type DeadLetterHearing struct {
+	follower DeadLetterFollower
+	streams  []string
+	dead     *DeadLetters
+	logf     func(string, ...any)
+	said     func(Hearing)
+	ctx      context.Context
+	since    time.Time
+
+	mu     sync.Mutex
+	stops  map[string]func()
+	why    map[string]string
+	trying bool
+	again  bool
+	told   *Hearing
+	wg     sync.WaitGroup
+}
+
+// NewDeadLetterHearing starts hearing the given-up asks of each language's build seat stream, in the background. said
+// is told each change of whether it hears (nil tells nothing); ctx ends it.
+func NewDeadLetterHearing(ctx context.Context, follower DeadLetterFollower, languages []string, dead *DeadLetters,
+	logf func(string, ...any), said func(Hearing)) *DeadLetterHearing {
+	h := &DeadLetterHearing{follower: follower, dead: dead, logf: logf, said: said, ctx: ctx, since: time.Now(),
+		stops: map[string]func(){}, why: map[string]string{}}
+	for _, l := range languages {
+		s := SeatStream(l)
+		h.streams = append(h.streams, s)
+		h.why[s] = "not tried yet"
+	}
+	if h.logf == nil {
+		h.logf = func(string, ...any) {}
+	}
+	h.Again()
+	return h
+}
+
+// Deaf says why a stream is not heard, or "" when every one is.
+func (h *DeadLetterHearing) Deaf() string {
+	h.mu.Lock()
+	defer h.mu.Unlock()
+	return h.deafLocked()
+}
+
+func (h *DeadLetterHearing) deafLocked() string {
+	var out []string
+	for _, s := range h.streams {
+		if why, deaf := h.why[s]; deaf {
+			out = append(out, s+": "+why)
+		}
+	}
+	if len(out) == 0 {
+		return ""
+	}
+	return DeadLettersStream + " is not heard, so an ask a build seat gives up on is waited for until its bound: " +
+		strings.Join(out, "; ")
+}
+
+// Again tries, in the background, every stream not heard yet; a call while it tries makes it try once more after.
+func (h *DeadLetterHearing) Again() {
+	h.mu.Lock()
+	if h.trying {
+		h.again = true
+		h.mu.Unlock()
+		return
+	}
+	h.trying = true
+	h.mu.Unlock()
+	h.wg.Add(1)
+	go func() {
+		defer h.wg.Done()
+		for {
+			h.try()
+			h.mu.Lock()
+			if !h.again || h.ctx.Err() != nil {
+				h.trying = false
+				h.mu.Unlock()
+				return
+			}
+			h.again = false
+			h.mu.Unlock()
+		}
+	}()
+}
+
+// Wait returns when no try runs.
+func (h *DeadLetterHearing) Wait() { h.wg.Wait() }
+
+// Stop ends every stream's following.
+func (h *DeadLetterHearing) Stop() {
+	h.mu.Lock()
+	defer h.mu.Unlock()
+	for s, stop := range h.stops {
+		stop()
+		delete(h.stops, s)
+	}
+}
+
+func (h *DeadLetterHearing) try() {
+	for _, s := range h.streams {
+		h.mu.Lock()
+		_, deaf := h.why[s]
+		h.mu.Unlock()
+		if !deaf || h.ctx.Err() != nil {
+			continue
+		}
+		stream := s
+		stop, err := h.follower.Follow(h.ctx, stream, h.since, func(g GivenUp) {
+			h.logf("the ask %s on %s was given up after %d deliveries at %s: the wait for its step ends", g.AskID, g.Stream,
+				g.Deliveries, g.At.Format(time.RFC3339))
+			h.dead.Heard(g)
+		})
+		h.mu.Lock()
+		if err != nil {
+			h.why[stream] = err.Error()
+			h.logf("the asks %s gives up on are not heard (%s of %s); tried again at the next reconnect or ask: %v",
+				stream, DeadLetterFilter(stream), DeadLettersStream, err)
+		} else {
+			delete(h.why, stream)
+			h.stops[stream] = stop
+			h.logf("the asks %s gives up on are heard from %s", stream, DeadLettersStream)
+		}
+		h.mu.Unlock()
+	}
+	h.tell()
+}
+
+// tell says the hearing when it changed.
+func (h *DeadLetterHearing) tell() {
+	h.mu.Lock()
+	why := h.deafLocked()
+	now := Hearing{Heard: why == "", Streams: append([]string(nil), h.streams...), Why: why, At: time.Now().UTC()}
+	same := h.told != nil && h.told.Heard == now.Heard && h.told.Why == now.Why
+	if !same {
+		h.told = &now
+	}
+	h.mu.Unlock()
+	if !same && h.said != nil {
+		h.said(now)
 	}
 }