diff --git a/internal/link/builds_nats.go b/internal/link/builds_nats.go index f95a2295..b2ea2151 100644 --- a/internal/link/builds_nats.go +++ b/internal/link/builds_nats.go @@ -362,6 +362,19 @@ func (m *natsMachine) Take(ctx context.Context, do func(context.Context, Build)) _ = msg.Nak() continue } + // **A duplicate of a check another builder is running is dropped** (issue 1649): an ask redelivered to this machine + // because its first holder did not say, in time, that it had it. The running key names that builder and this same + // ask; running it a second time would spend a second place on one check. + if why := m.runningElsewhere(ctx, request); why != "" { + fmt.Fprintf(os.Stderr, "%s is handed out again while another builder runs it and is dropped: %s\n", request.ID, why) + _ = 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. @@ -371,11 +384,95 @@ 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 what is read from a pull without asking the server for more: the messages it already holds. +type heldSource interface { + NextMsg(timeout time.Duration) (*nats.Msg, error) +} + +// 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, 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 after its fetch gave up 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 { + break + } + hand(msg) + n++ + } + } + 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. +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} { + select { + case <-done: + return + case <-time.After(at - last): + } + last = at + 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 { + if r.Check == nil { + return "" + } + running, ok, err := ReadRunning(ctx, m.js.Conn(), r) + if err != nil { + fmt.Fprintf(os.Stderr, "could not read whether %s is running elsewhere, so it runs: %v\n", r.ID, err) + return "" + } + return duplicateOf(running, ok, r, m.on) +} + +// 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) { + return fmt.Sprintf("%s is running it as %s since %s", running.Builder, running.Build, running.Started.Local().Format("15:04:05")) + } + return "" +} + // bind binds a queue's worker, which exists and is not created here. // // **The consumer's own filter, not the one subject this machine cares about.** The client checks what is asked for diff --git a/internal/link/checks_nats_integration_test.go b/internal/link/checks_nats_integration_test.go index 71959b63..7afb3d07 100644 --- a/internal/link/checks_nats_integration_test.go +++ b/internal/link/checks_nats_integration_test.go @@ -510,3 +510,70 @@ func TestAnUnacknowledgedCheckGoesToASecondBuilder(t *testing.T) { stop() } } + +// An ask handed out a second time to another builder while the first still runs it is acknowledged and not run: the +// running key names the other builder and this same ask (issue 1649). +func TestAnAskHandedOutAgainWhileAnotherBuilderRunsItIsNotRunTwice(t *testing.T) { + js := aBusWithTheBuildRole(t) + aKeyBucket(t, js) + aRunningBucket(t, js, 2*time.Minute) + ctx, stop := context.WithCancel(t.Context()) + defer stop() + asks := BuildsOn(js, TheBuildMachine) + running := make(chan string, 4) + release := make(chan struct{}) + a := MachineOverNATS(js, "builder-a") + defer a.Close() + go func() { + _ = a.Take(ctx, func(ctx context.Context, work Build) { + running <- work.Request().ID + <-release + _ = work.Announce(ctx, BuildResult{ID: work.Request().ID, On: "builder-a", Failed: "no"}) + _ = work.Done() + }) + }() + check := aPullCheck("build-twice", 7, "head-7-000000000", "trunk-1-000000", false) + if err := asks.Ask(ctx, check); err != nil { + t.Fatal(err) + } + select { + case <-running: + case <-time.After(10 * time.Second): + t.Fatal("builder a took nothing") + } + // The bus's redelivery, as it reaches the other machine: the same ask, the same id. + if err := asks.Ask(ctx, check); 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.Done() + }) + }() + select { + case id := <-took: + t.Fatalf("builder b ran %s, which builder a is running", id) + case <-time.After(3 * time.Second): + } + if n := len(queueOf(t, js).Of(AskWaiting)); n != 0 { + t.Errorf("%d request(s) still wait; builder b was to acknowledge the duplicate", n) + } + 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) + } +} diff --git a/internal/link/held_asks_test.go b/internal/link/held_asks_test.go new file mode 100644 index 00000000..6f84a620 --- /dev/null +++ b/internal/link/held_asks_test.go @@ -0,0 +1,61 @@ +package link + +import ( + "errors" + "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) NextMsg(time.Duration) (*nats.Msg, error) { + if len(f.msgs) == 0 { + return nil, nats.ErrTimeout + } + m := f.msgs[0] + f.msgs = f.msgs[1:] + return m, nil +} + +type brokenHeld struct{} + +func (brokenHeld) NextMsg(time.Duration) (*nats.Msg, error) { return nil, errors.New("closed") } + +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) }) + 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) + } +} + +func TestAnAskAnotherBuilderIsRunningIsADuplicateAndYoursIsNot(t *testing.T) { + r := aPullCheck("build-1", 423, "head-4771ee53aa", "trunk-0123456789", false) + run := CheckRunning{Head: r.Ref, Trunk: "trunk-0123456789", Builder: "shanks", Build: "build-1", Started: time.Now()} + for name, c := range map[string]struct { + running CheckRunning + found bool + me string + dup bool + }{ + "another builder runs this very ask": {run, true, "ace", true}, + "this builder runs it itself": {run, true, "shanks", false}, + "nobody runs it": {CheckRunning{}, false, "ace", false}, + "another ask of the same head": {CheckRunning{Head: r.Ref, Trunk: "trunk-0123456789", Builder: "shanks", Build: "build-0"}, true, "ace", false}, + "the same ask at another head": {CheckRunning{Head: "head-other-0000", Trunk: "trunk-0123456789", Builder: "shanks", Build: "build-1"}, true, "ace", false}, + } { + if got := duplicateOf(c.running, c.found, r, c.me) != ""; got != c.dup { + t.Errorf("%s: duplicate = %v; want %v", name, got, c.dup) + } + } +}