diff --git a/internal/link/builds_nats.go b/internal/link/builds_nats.go index a3d4a15d..b901e9fe 100644 --- a/internal/link/builds_nats.go +++ b/internal/link/builds_nats.go @@ -370,11 +370,6 @@ func (m *natsMachine) Take(ctx context.Context, do func(context.Context, Build)) _ = msg.Ack() continue } - // **Nothing else is held while this builds** (issue 1649): an ask the server delivered to another queue's pull in - // the moment its fetch gave up sits in that pull's buffer, gets no heartbeat, and is handed to another machine after - // the ack wait while this one would still run it later. Handed back now and in the first seconds of the build. - handBackHeld(m.heldSources(), handBack) - stopHanding := m.handBackWhileBuilding(handBack) // A build outlives the acknowledgement window many times over; said while it runs, // as the controller says it for its own long handlers, so the server neither hands // the ask to a second machine nor counts the wait against its deliveries. @@ -384,95 +379,11 @@ func (m *natsMachine) Take(ctx context.Context, do func(context.Context, Build)) // deleted when it ends, whatever the outcome. stopRunning := keepRunning(ctx, m.js.Conn(), request, m.on, heartbeatWhileWorking) do(ctx, build) - stopHanding() stopRunning() close(working) } } -// heldSource is a pull that may hold a message the server already delivered to it. -type heldSource interface { - // 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. -func (m *natsMachine) heldSources() []heldSource { - var out []heldSource - for _, q := range m.subs { - if q.bound && q.sub != nil { - out = append(out, pullHeld{q.sub}) - } - } - return out -} - -// handBack gives a message nobody here will start back to the queue, for another holder. -func handBack(msg *nats.Msg) { - if err := msg.Nak(); err != nil { - 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 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 i := 0; i < 32; i++ { - msg := s.Held() - if msg == nil { - break - } - hand(msg) - n++ - } - } - return n -} - -// 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) - tick := time.NewTicker(heldLookEvery) - defer tick.Stop() - for { - select { - case <-done: - return - case <-tick.C: - handBackHeld(m.heldSources(), hand) - } - } - }() - return func() { close(done); <-finished } -} - // runningElsewhere says why an ask need not run here, or nothing: the bucket of the checks running says another builder // runs this very ask (its id, its head and its trunk commit). A key that cannot be read runs the check. func (m *natsMachine) runningElsewhere(ctx context.Context, r BuildRequest) string { @@ -493,6 +404,7 @@ 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 { + // Seen is the server's time and runningFreshFor is judged against this builder's clock: a skew over about 35 s would matter. // 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 { diff --git a/internal/link/checks_nats.go b/internal/link/checks_nats.go index d0895699..b720bbf4 100644 --- a/internal/link/checks_nats.go +++ b/internal/link/checks_nats.go @@ -253,11 +253,25 @@ func watchPullIn(ctx context.Context, buckets []string, open func(ctx context.Co // checkTakeWait is how long a taking loop waits on one queue for an ask that is not there, and takeIdleWait how long it // waits on the first when all three are empty. +// +// **A fetch's pull request must end before the fetch does** (issue 1649). nats.go gives the server-side request an expiry a tenth +// shorter than the fetch's wait (fetchExpiryOf), and the server then answers with a status message at the expiry; a wait so +// short that the answer can land after the fetch gave up left a status message in the pull's buffer, and anything a request +// could still deliver would be held by a machine that is busy. Each wait leaves at least minFetchMargin between the two. const ( - checkTakeWait = 30 * time.Millisecond + checkTakeWait = 100 * time.Millisecond takeIdleWait = time.Second ) +// minFetchMargin is the least time between a fetch's server-side expiry and the end of its wait. +const minFetchMargin = 10 * time.Millisecond + +// fetchExpiryOf is the expiry nats.go gives the pull request of a fetch that waits this long: the wait less a tenth of it, or +// less five seconds. +func fetchExpiryOf(wait time.Duration) time.Duration { + return wait - min(wait/10, 5*time.Second) +} + // CheckRunning is what a builder says of the check it is running, in the bucket of the checks running (novox/hq ADR 0406 // ยง3): the head and trunk commit it checks, the builder and its build. The key is the pull request's (RunningKeyOf). type CheckRunning struct { diff --git a/internal/link/checks_nats_integration_test.go b/internal/link/checks_nats_integration_test.go index a1cdc255..5e535632 100644 --- a/internal/link/checks_nats_integration_test.go +++ b/internal/link/checks_nats_integration_test.go @@ -566,9 +566,9 @@ func TestAnAskHandedOutAgainWhileAnotherBuilderRunsItIsNotRunTwice(t *testing.T) close(release) } -// 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) { +// A busy machine fetches nothing, so no ask waiting in the queue is taken by it, whatever status message sits in its pulls +// (issue 1649). A pull request of our own that expires at once leaves such a message in the other checks' pull. +func TestABusyMachineTakesNoWaitingAskWhateverItsPullsHold(t *testing.T) { js := aBusWithTheBuildRole(t) aKeyBucket(t, js) aRunningBucket(t, js, 2*time.Minute) @@ -595,9 +595,6 @@ func TestAnAskDeliveredToABusyMachinesIdlePullIsHandedBackAtOnce(t *testing.T) { case <-time.After(10 * time.Second): t.Fatal("builder a took nothing") } - if err := asks.Ask(ctx, aPullCheck("held-check", 9, "head-9-000000000", "trunk-1-000000", false)); err != nil { - t.Fatal(err) - } var other *takeQueue for _, q := range a.subs { if q.kind == queueOther && q.bound { @@ -607,28 +604,29 @@ func TestAnAskDeliveredToABusyMachinesIdlePullIsHandedBackAtOnce(t *testing.T) { if other == nil { t.Fatal("builder a has no pull on the other checks") } + // The queue is empty, so this request is answered at its expiry with a status message that lands in the pull. reply := strings.TrimSuffix(other.sub.Subject, "*") + "late" if err := js.Conn().PublishRequest("$JS.API.CONSUMER.MSG.NEXT."+other.worker.Stream+"."+other.worker.Name, reply, - []byte(`{"batch":1,"expires":5000000000}`)); err != nil { + []byte(`{"batch":1,"expires":1000000}`)); err != nil { t.Fatal(err) } - b := MachineOverNATS(js, "builder-b") - defer b.Close() - took := make(chan string, 4) - go func() { - _ = b.Take(ctx, func(ctx context.Context, work Build) { - took <- work.Request().ID - _ = work.Announce(ctx, BuildResult{ID: work.Request().ID, On: "builder-b", Failed: "no"}) - _ = work.Done() - }) - }() - select { - case id := <-took: - if id != "held-check" { - t.Fatalf("builder b took %s", id) + time.Sleep(500 * time.Millisecond) + for i, id := range []string{"wait-1", "wait-2"} { + if err := asks.Ask(ctx, aPullCheck(id, 20+i, "head-x-000000000", "trunk-1-000000", false)); err != nil { + t.Fatal(err) } - case <-time.After(15 * time.Second): - t.Fatal("the ask held by busy builder a was not handed back; it would wait out the 60 s ack wait") + } + time.Sleep(5 * time.Second) + info, err := js.Context().ConsumerInfo(other.worker.Stream, other.worker.Name) + if err != nil { + t.Fatal(err) + } + if info.NumAckPending != 0 || info.NumRedelivered != 0 || info.Delivered.Consumer != 0 { + t.Fatalf("a busy machine took from the queue: %d delivered, %d pending, %d redelivered", info.Delivered.Consumer, + info.NumAckPending, info.NumRedelivered) + } + if n := len(queueOf(t, js).Of(AskWaiting)); n != 2 { + t.Fatalf("%d asks wait; want the 2 nobody took", n) } close(release) } diff --git a/internal/link/held_asks_test.go b/internal/link/held_asks_test.go index 0e688904..9119dfcc 100644 --- a/internal/link/held_asks_test.go +++ b/internal/link/held_asks_test.go @@ -3,38 +3,18 @@ package link import ( "testing" "time" - - "github.com/nats-io/nats.go" ) // An ask a machine holds but does not start gets no heartbeat, and the bus hands it to another machine after the ack // wait, so two machines run one check (issue 1649). -type fakeHeld struct{ msgs []*nats.Msg } - -func (f *fakeHeld) Held() *nats.Msg { - if len(f.msgs) == 0 { - return nil - } - m := f.msgs[0] - f.msgs = f.msgs[1:] - return m -} - -type emptyHeld struct{} - -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, 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) - } - if n := handBackHeld([]heldSource{a, b}, func(*nats.Msg) { t.Fatal("handed back what was not held") }); n != 0 { - t.Fatalf("a second pass handed back %d", n) +// A fetch's server-side pull request ends clearly before the fetch does, so the server's answer and any message it could still +// deliver come inside the fetch and never into a buffer nobody reads while the machine builds. +func TestAFetchsPullRequestEndsClearlyBeforeTheFetchDoes(t *testing.T) { + for name, wait := range map[string]time.Duration{"a check queue": checkTakeWait, "the idle wait": takeIdleWait} { + if margin := wait - fetchExpiryOf(wait); margin < minFetchMargin { + t.Errorf("%s: the pull request ends %v before a fetch of %v does; want at least %v", name, margin, wait, minFetchMargin) + } } }