diff --git a/internal/broker/seattraffic.go b/internal/broker/seattraffic.go index be2852c4..eea8e19c 100644 --- a/internal/broker/seattraffic.go +++ b/internal/broker/seattraffic.go @@ -1,6 +1,9 @@ package broker -import "sort" +import ( + "fmt" + "sort" +) // What a module may say and take on the seats it holds, uses and watches, beyond tools (novox/hq ADR // 0259 §3, to-be 46 §10). @@ -24,6 +27,10 @@ import "sort" // And one read: **a holder's records**, a bucket the seat names, read by each user under its own name only // (`$KV...>`), so an asker reads the state of its own asks and no other asker's. +// BuildAskMaxDeliver bounds the deliveries of a build seat's ask (issue 1907): mesh-sdk delivery.AskMaxDeliver, written +// here as the broker mirrors the catalogue rather than importing it. +const BuildAskMaxDeliver = 5 + // KindedBenches are the seats that may be kinded (ADR 0234 §2): making another is a decision, recorded. var KindedBenches = map[string]bool{"channel": true, "intake": true} @@ -190,10 +197,10 @@ func SeatTrafficObjects(users []Principal) ([]Stream, []Consumer) { consumers := map[string]Consumer{} add := func(module string, holds []Seat) { for _, w := range SeatTrafficOf(module, holds, nil, nil).Workers { - seat := "" + seat, byKind := "", false for _, s := range holds { if seatStreamName(s.Name) == w.Stream { - seat = s.Name + seat, byKind = s.Name, s.PulledByKind } } streams[w.Stream] = Stream{ @@ -203,7 +210,7 @@ func SeatTrafficObjects(users []Principal) ([]Stream, []Consumer) { MaxAge: 7 * 24 * 60 * 60, Why: "work submitted to the " + seat + " seat; its holders take it, each kind its own, and it queues while nobody does", } - consumers[w.Consumer] = Consumer{ + c := Consumer{ Name: w.Consumer, Stream: w.Stream, Filters: []string{w.Filter}, @@ -214,6 +221,15 @@ func SeatTrafficObjects(users []Principal) ([]Stream, []Consumer) { Why: module + " holds " + seat + "; it pulls one ask at a time and acknowledges once it has " + "recorded it, so a crash redelivers rather than loses", } + if byKind { + // A build seat's ask is offered a bounded number of times (issue 1907): the holder that takes it past + // mesh-sdk delivery.AskRuns ends it failed saying it is poison, and the bound stops a run that kills + // every node it reaches even when that holder dies too. + c.MaxDeliver = BuildAskMaxDeliver + c.Why = module + " holds " + seat + "; each holder pulls an ask when it has a free slot, and an ask that " + + "no run ends is offered at most " + fmt.Sprint(BuildAskMaxDeliver) + " times, its step failed as poison" + } + consumers[w.Consumer] = c } } for _, p := range users { diff --git a/internal/broker/seattraffic_test.go b/internal/broker/seattraffic_test.go index 26295f02..cfe8218b 100644 --- a/internal/broker/seattraffic_test.go +++ b/internal/broker/seattraffic_test.go @@ -428,6 +428,12 @@ func TestTheBuilderMayAskABuildSeatAndASessionMayNot(t *testing.T) { if len(consumers) != 3 { t.Errorf("consumers = %+v, want the three kinds' workers", consumers) } + // 1907: a build ask is offered a bounded number of times, past the holder's poison count (mesh-sdk AskRuns 3). + for _, c := range consumers { + if c.MaxDeliver != BuildAskMaxDeliver || BuildAskMaxDeliver <= 3 { + t.Errorf("%s offers an ask %d times, want %d", c.Name, c.MaxDeliver, BuildAskMaxDeliver) + } + } if _, made := HolderConsumerFor("shanks", "mesh-build-go", DeclaredSeat{Name: "node-build-go", Accepts: buildSeat().Accepts, PulledByKind: true}); made { t.Error("a catch-all worker would overlap the kinds' workers on the work queue") }