diff --git a/go/buscompose/buscompose_test.go b/go/buscompose/buscompose_test.go index 1a685ed..e7ff21c 100644 --- a/go/buscompose/buscompose_test.go +++ b/go/buscompose/buscompose_test.go @@ -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) diff --git a/go/buscompose/seattraffic.go b/go/buscompose/seattraffic.go index d7b1430..e89b208 100644 --- a/go/buscompose/seattraffic.go +++ b/go/buscompose/seattraffic.go @@ -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 + ".*", } } diff --git a/go/delivery/buildrun/asks_test.go b/go/delivery/buildrun/asks_test.go index 44281b1..f5ea8c3 100644 --- a/go/delivery/buildrun/asks_test.go +++ b/go/delivery/buildrun/asks_test.go @@ -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) } diff --git a/go/delivery/buildrun/lostask.go b/go/delivery/buildrun/lostask.go index 52f7ee8..1e70245 100644 --- a/go/delivery/buildrun/lostask.go +++ b/go/delivery/buildrun/lostask.go @@ -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) } }