From 650274d7000cfa01f41f56bda675566c071f456f Mon Sep 17 00:00:00 2001 From: jochen Date: Wed, 21 Oct 2026 16:58:39 +0200 Subject: [PATCH] Read a pull's held asks with Fetch, hand them back all through a build, and drop a duplicate only while its builder's key is fresh (issue 1649) --- internal/link/builds_nats.go | 58 ++++++++++++++----- internal/link/checks_nats.go | 1 + internal/link/checks_nats_integration_test.go | 15 +---- internal/link/held_asks_test.go | 13 ++--- 4 files changed, 51 insertions(+), 36 deletions(-) diff --git a/internal/link/builds_nats.go b/internal/link/builds_nats.go index b2ea2151..a3d4a15d 100644 --- a/internal/link/builds_nats.go +++ b/internal/link/builds_nats.go @@ -390,9 +390,28 @@ func (m *natsMachine) Take(ctx context.Context, do func(context.Context, Build)) } } -// heldSource is what is read from a pull without asking the server for more: the messages it already holds. +// heldSource is a pull that may hold a message the server already delivered to it. type heldSource interface { - NextMsg(timeout time.Duration) (*nats.Msg, error) + // Held takes the message the pull holds, or nil when it holds none. + Held() *nats.Msg +} + +// pullHeld is a pull subscription as a heldSource. NextMsg is refused on a pull subscription (ErrTypeSubscription), so the +// buffer is read with Fetch, which returns what the subscription holds before it asks the server for anything; it is only +// called when Pending says something is held. +type pullHeld struct{ sub *nats.Subscription } + +func (p pullHeld) Held() *nats.Msg { + if pending, _, err := p.sub.Pending(); err != nil || pending == 0 { + return nil + } + ctx, end := context.WithTimeout(context.Background(), 20*time.Millisecond) + defer end() + got, err := p.sub.Fetch(1, nats.Context(ctx)) + if err != nil || len(got) == 0 { + return nil + } + return got[0] } // heldSources are the machine's pulls, which may hold a message already delivered to them. @@ -400,7 +419,7 @@ func (m *natsMachine) heldSources() []heldSource { var out []heldSource for _, q := range m.subs { if q.bound && q.sub != nil { - out = append(out, q.sub) + out = append(out, pullHeld{q.sub}) } } return out @@ -412,16 +431,16 @@ func handBack(msg *nats.Msg) { fmt.Fprintf(os.Stderr, "could not hand a held ask back to the queue: %v\n", err) return } - fmt.Fprintln(os.Stderr, "an ask delivered to this machine's pull after its fetch gave up is handed back, not held") + fmt.Fprintln(os.Stderr, "an ask delivered to this machine's pull while it was not fetching is handed back, not held") } // handBackHeld hands back every message the sources hold, none waited for, and says how many. func handBackHeld(sources []heldSource, hand func(*nats.Msg)) int { n := 0 for _, s := range sources { - for { - msg, err := s.NextMsg(0) - if err != nil || msg == nil { + for i := 0; i < 32; i++ { + msg := s.Held() + if msg == nil { break } hand(msg) @@ -431,21 +450,24 @@ func handBackHeld(sources []heldSource, hand func(*nats.Msg)) int { return n } -// handBackWhileBuilding hands back what a pull is delivered in the first seconds of a build, the time a late delivery can -// still arrive (a pull request lives a few milliseconds past its fetch). The stop ends it and waits for it. +// heldLookEvery is how often a machine that is building looks for an ask delivered to one of its pulls. +const heldLookEvery = time.Second + +// handBackWhileBuilding hands back what a pull is delivered during a build, looking every second until the stop, which +// ends it and waits for it. func (m *natsMachine) handBackWhileBuilding(hand func(*nats.Msg)) (stop func()) { done, finished := make(chan struct{}), make(chan struct{}) go func() { defer close(finished) - last := time.Duration(0) - for _, at := range []time.Duration{100 * time.Millisecond, 500 * time.Millisecond, 2 * time.Second} { + tick := time.NewTicker(heldLookEvery) + defer tick.Stop() + for { select { case <-done: return - case <-time.After(at - last): + case <-tick.C: + handBackHeld(m.heldSources(), hand) } - last = at - handBackHeld(m.heldSources(), hand) } }() return func() { close(done); <-finished } @@ -465,9 +487,15 @@ func (m *natsMachine) runningElsewhere(ctx context.Context, r BuildRequest) stri return duplicateOf(running, ok, r, m.on) } +// runningFreshFor is how recent a running key's last heartbeat must be for its builder to be taken as alive: two missed +// heartbeats of ten seconds are not yet a dead builder, and a redelivery after a crash comes a minute after the last one. +const runningFreshFor = 25 * time.Second + // duplicateOf says why an ask is a second hand-out of one another builder is running, or nothing. func duplicateOf(running CheckRunning, found bool, r BuildRequest, me string) string { - if found && running.Build == r.ID && running.Builder != me && running.Same(r) { + // A key not renewed for runningFreshFor is a builder that has gone: its ask is redelivered after the ack wait, and + // that is the recovery, not a duplicate (the key itself lives two minutes). + if found && running.Build == r.ID && running.Builder != me && running.Same(r) && time.Since(running.Seen) < runningFreshFor { return fmt.Sprintf("%s is running it as %s since %s", running.Builder, running.Build, running.Started.Local().Format("15:04:05")) } return "" diff --git a/internal/link/checks_nats.go b/internal/link/checks_nats.go index 3d472515..d0895699 100644 --- a/internal/link/checks_nats.go +++ b/internal/link/checks_nats.go @@ -321,6 +321,7 @@ func ReadRunning(ctx context.Context, conn *nats.Conn, r BuildRequest) (CheckRun if err := json.Unmarshal(e.Value(), &running); err != nil { return CheckRunning{}, false, nil // a key of a builder from before this means no match } + running.Seen = e.Created() return running, true, nil } diff --git a/internal/link/checks_nats_integration_test.go b/internal/link/checks_nats_integration_test.go index 6ba9b83c..a1cdc255 100644 --- a/internal/link/checks_nats_integration_test.go +++ b/internal/link/checks_nats_integration_test.go @@ -3,10 +3,10 @@ package link import ( - "strings" "context" "encoding/json" "reflect" + "strings" "testing" "time" @@ -566,19 +566,6 @@ func TestAnAskHandedOutAgainWhileAnotherBuilderRunsItIsNotRunTwice(t *testing.T) close(release) } -// A real pull answers NextMsg from its buffer without asking the server, and a pull holding nothing hands back nothing. -func TestARealPullHoldingNothingHandsBackNothing(t *testing.T) { - js := aBusWithTheBuildRole(t) - w := broker.CheckWorkers()[0] - sub, err := js.Context().PullSubscribe(w.Filters[0], w.Name, nats.Bind(w.Stream, w.Name), nats.ManualAck()) - if err != nil { - t.Fatal(err) - } - if n := handBackHeld([]heldSource{sub}, func(*nats.Msg) { t.Fatal("held nothing") }); n != 0 { - t.Fatalf("handed back %d", n) - } -} - // An ask the server delivers to one of a busy machine's pulls outside a fetch (here by a pull request of our own, as one // that outlived its fetch) is handed back at once for another machine, not held until the ack wait (issue 1649). func TestAnAskDeliveredToABusyMachinesIdlePullIsHandedBackAtOnce(t *testing.T) { diff --git a/internal/link/held_asks_test.go b/internal/link/held_asks_test.go index 47947a21..0e688904 100644 --- a/internal/link/held_asks_test.go +++ b/internal/link/held_asks_test.go @@ -1,7 +1,6 @@ package link import ( - "errors" "testing" "time" @@ -13,24 +12,24 @@ import ( type fakeHeld struct{ msgs []*nats.Msg } -func (f *fakeHeld) NextMsg(time.Duration) (*nats.Msg, error) { +func (f *fakeHeld) Held() *nats.Msg { if len(f.msgs) == 0 { - return nil, nats.ErrTimeout + return nil } m := f.msgs[0] f.msgs = f.msgs[1:] - return m, nil + return m } -type brokenHeld struct{} +type emptyHeld struct{} -func (brokenHeld) NextMsg(time.Duration) (*nats.Msg, error) { return nil, errors.New("closed") } +func (emptyHeld) Held() *nats.Msg { return nil } func TestEveryMessageAPullHoldsIsHandedBackBeforeABuildStarts(t *testing.T) { a := &fakeHeld{msgs: []*nats.Msg{{Subject: "a1"}, {Subject: "a2"}}} b := &fakeHeld{msgs: []*nats.Msg{{Subject: "b1"}}} var back []string - n := handBackHeld([]heldSource{a, brokenHeld{}, b}, func(m *nats.Msg) { back = append(back, m.Subject) }) + n := handBackHeld([]heldSource{a, emptyHeld{}, b}, func(m *nats.Msg) { back = append(back, m.Subject) }) if n != 3 || len(back) != 3 || back[0] != "a1" || back[2] != "b1" { t.Fatalf("handed back %d %v; want a1 a2 b1", n, back) }