diff --git a/internal/link/builds_nats.go b/internal/link/builds_nats.go index f95a2295..b901e9fe 100644 --- a/internal/link/builds_nats.go +++ b/internal/link/builds_nats.go @@ -362,6 +362,14 @@ 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 + } // 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. @@ -376,6 +384,35 @@ func (m *natsMachine) Take(ctx context.Context, do func(context.Context, Build)) } } +// 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) +} + +// 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 { + // 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 { + 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.go b/internal/link/checks_nats.go index 176c70ec..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 { @@ -268,6 +282,8 @@ type CheckRunning struct { // Started is when the check started, kept on every heartbeat: the entry's own time is the last heartbeat's, and a // reader that asks how long a check has run reads this (novox/hq ADR 0433 §6). Started time.Time `json:"started"` + // Seen is when the key was last put, which is the last heartbeat: read from the entry, never written (issue 1649). + Seen time.Time `json:"-"` } // RunningKeyOf is the key a check is said running under: its pull request's tokens, or `group.` for a group's @@ -319,6 +335,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 71959b63..5e535632 100644 --- a/internal/link/checks_nats_integration_test.go +++ b/internal/link/checks_nats_integration_test.go @@ -6,6 +6,7 @@ import ( "context" "encoding/json" "reflect" + "strings" "testing" "time" @@ -510,3 +511,122 @@ 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 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) + ctx, stop := context.WithCancel(t.Context()) + defer stop() + asks := BuildsOn(js, TheBuildMachine) + busy := make(chan string, 4) + release := make(chan struct{}) + a := MachineOverNATS(js, "builder-a").(*natsMachine) + defer a.Close() + go func() { + _ = a.Take(ctx, func(ctx context.Context, work Build) { + busy <- work.Request().ID + <-release + _ = work.Announce(ctx, BuildResult{ID: work.Request().ID, On: "builder-a", Failed: "no"}) + _ = work.Done() + }) + }() + if err := asks.Ask(ctx, BuildRequest{ID: "busy", Repository: "/d"}); err != nil { + t.Fatal(err) + } + select { + case <-busy: + case <-time.After(10 * time.Second): + t.Fatal("builder a took nothing") + } + var other *takeQueue + for _, q := range a.subs { + if q.kind == queueOther && q.bound { + other = q + } + } + 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":1000000}`)); err != nil { + t.Fatal(err) + } + 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) + } + } + 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 new file mode 100644 index 00000000..9119dfcc --- /dev/null +++ b/internal/link/held_asks_test.go @@ -0,0 +1,43 @@ +package link + +import ( + "testing" + "time" +) + +// 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). + +// 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) + } + } +} + +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().Add(-time.Minute), Seen: time.Now().Add(-4 * time.Second)} + dead := run + dead.Seen = time.Now().Add(-60 * time.Second) + 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}, + "its builder stopped renewing the key a minute ago": {dead, true, "ace", false}, + "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) + } + } +}