From 14e0808fd441dd891dfaca206f173d3383e85492 Mon Sep 17 00:00:00 2001 From: Seth Boyles Date: Mon, 2 Nov 2026 15:07:18 -0700 Subject: [PATCH] Back off job polling from 250ms up to the polling interval Job polling checked the job once immediately and then slept a fixed polling interval (3s) between checks. Short asynchronous jobs, such as app and route deletes, are rarely complete at the first check, so every such command took at least 3s even when the job finished in well under a second. Poll at 250ms, 750ms and 1.75s after the start, then on the existing polling-interval grid (3s, 6s, ...), so every check today's CLI makes still happens and no job is detected later than before. The event stream now closes right after the terminal event instead of sleeping one more interval first. Timeout checks are unchanged. Add Sleep to the ccv3 Clock so the delay sequence can be tested deterministically. ai-assisted=yes AI-Tool-Used: Claude, Codex AI-Tool-Use-Level: High AI-Code-Category: Production Co-Authored-By: Claude Opus 5.5 --- .../ccv3/ccv3fakes/fake_clock.go | 64 ++++- api/cloudcontroller/ccv3/clock.go | 1 + .../ccv3/internal/real_time.go | 4 + api/cloudcontroller/ccv3/job.go | 24 +- api/cloudcontroller/ccv3/job_test.go | 244 ++++++++++++++---- 5 files changed, 286 insertions(+), 51 deletions(-) diff --git a/api/cloudcontroller/ccv3/ccv3fakes/fake_clock.go b/api/cloudcontroller/ccv3/ccv3fakes/fake_clock.go index 204be49973a..31ad22157ee 100644 --- a/api/cloudcontroller/ccv3/ccv3fakes/fake_clock.go +++ b/api/cloudcontroller/ccv3/ccv3fakes/fake_clock.go @@ -11,23 +11,30 @@ import ( type FakeClock struct { NowStub func() time.Time nowMutex sync.RWMutex - nowArgsForCall []struct { - } - nowReturns struct { + nowArgsForCall []struct{} + nowReturns struct { result1 time.Time } nowReturnsOnCall map[int]struct { result1 time.Time } + SleepStub func(time.Duration) + sleepMutex sync.RWMutex + sleepArgsForCall []FakeClockSleepArgs invocations map[string][][]interface{} + callOrder []string invocationsMutex sync.RWMutex } +// FakeClockSleepArgs holds the arguments of one call to Sleep. +type FakeClockSleepArgs struct { + Arg1 time.Duration +} + func (fake *FakeClock) Now() time.Time { fake.nowMutex.Lock() ret, specificReturn := fake.nowReturnsOnCall[len(fake.nowArgsForCall)] - fake.nowArgsForCall = append(fake.nowArgsForCall, struct { - }{}) + fake.nowArgsForCall = append(fake.nowArgsForCall, struct{}{}) stub := fake.NowStub fakeReturns := fake.nowReturns fake.recordInvocation("Now", []interface{}{}) @@ -76,6 +83,44 @@ func (fake *FakeClock) NowReturnsOnCall(i int, result1 time.Time) { }{result1} } +func (fake *FakeClock) Sleep(arg1 time.Duration) { + fake.sleepMutex.Lock() + fake.sleepArgsForCall = append(fake.sleepArgsForCall, FakeClockSleepArgs{arg1}) + stub := fake.SleepStub + fake.recordInvocation("Sleep", []interface{}{arg1}) + fake.sleepMutex.Unlock() + if stub != nil { + stub(arg1) + } +} + +func (fake *FakeClock) SleepCallCount() int { + fake.sleepMutex.RLock() + defer fake.sleepMutex.RUnlock() + return len(fake.sleepArgsForCall) +} + +func (fake *FakeClock) SleepCalls(stub func(time.Duration)) { + fake.sleepMutex.Lock() + defer fake.sleepMutex.Unlock() + fake.SleepStub = stub +} + +func (fake *FakeClock) SleepArgsForCall(i int) time.Duration { + fake.sleepMutex.RLock() + defer fake.sleepMutex.RUnlock() + argsForCall := fake.sleepArgsForCall[i] + return argsForCall.Arg1 +} + +func (fake *FakeClock) SleepArgs() []FakeClockSleepArgs { + fake.sleepMutex.RLock() + defer fake.sleepMutex.RUnlock() + args := make([]FakeClockSleepArgs, len(fake.sleepArgsForCall)) + copy(args, fake.sleepArgsForCall) + return args +} + func (fake *FakeClock) Invocations() map[string][][]interface{} { fake.invocationsMutex.RLock() defer fake.invocationsMutex.RUnlock() @@ -86,9 +131,18 @@ func (fake *FakeClock) Invocations() map[string][][]interface{} { return copiedInvocations } +func (fake *FakeClock) CallOrder() []string { + fake.invocationsMutex.RLock() + defer fake.invocationsMutex.RUnlock() + order := make([]string, len(fake.callOrder)) + copy(order, fake.callOrder) + return order +} + func (fake *FakeClock) recordInvocation(key string, args []interface{}) { fake.invocationsMutex.Lock() defer fake.invocationsMutex.Unlock() + fake.callOrder = append(fake.callOrder, key) if fake.invocations == nil { fake.invocations = map[string][][]interface{}{} } diff --git a/api/cloudcontroller/ccv3/clock.go b/api/cloudcontroller/ccv3/clock.go index b8f8c71ce2d..02437f58801 100644 --- a/api/cloudcontroller/ccv3/clock.go +++ b/api/cloudcontroller/ccv3/clock.go @@ -6,4 +6,5 @@ import "time" type Clock interface { Now() time.Time + Sleep(time.Duration) } diff --git a/api/cloudcontroller/ccv3/internal/real_time.go b/api/cloudcontroller/ccv3/internal/real_time.go index e759d1fae7f..84db6c542cc 100644 --- a/api/cloudcontroller/ccv3/internal/real_time.go +++ b/api/cloudcontroller/ccv3/internal/real_time.go @@ -7,3 +7,7 @@ type RealTime struct{} func (RealTime) Now() time.Time { return time.Now() } + +func (RealTime) Sleep(d time.Duration) { + time.Sleep(d) +} diff --git a/api/cloudcontroller/ccv3/job.go b/api/cloudcontroller/ccv3/job.go index a1346a7e8cd..63655286614 100644 --- a/api/cloudcontroller/ccv3/job.go +++ b/api/cloudcontroller/ccv3/job.go @@ -98,6 +98,24 @@ func (client *Client) PollJob(jobURL JobURL) (Warnings, error) { return client.PollJobForState(jobURL, constant.JobComplete) } +const initialJobPollingDelay = 250 * time.Millisecond + +func jobPollingDelay(elapsed, interval time.Duration) time.Duration { + if interval <= 0 { + return interval + } + + next := (elapsed/interval + 1) * interval + for r := initialJobPollingDelay; r < interval; r = 2*r + initialJobPollingDelay { + if r > elapsed { + next = r + break + } + } + + return next - elapsed +} + func (client *Client) PollJobForState(jobURL JobURL, state constant.JobState) (Warnings, error) { if jobURL == "" { return nil, nil @@ -137,7 +155,7 @@ func (client *Client) PollJobForState(jobURL JobURL, state constant.JobState) (W return allWarnings, nil } - time.Sleep(client.jobPollingInterval) + client.clock.Sleep(jobPollingDelay(client.clock.Now().Sub(startTime), client.jobPollingInterval)) } return allWarnings, ccerror.JobTimeoutError{ @@ -191,7 +209,9 @@ func (client *Client) PollJobToEventStream(jobURL JobURL) chan PollJobEvent { } stream <- event - time.Sleep(client.jobPollingInterval) + if !end { + client.clock.Sleep(jobPollingDelay(client.clock.Now().Sub(startTime), client.jobPollingInterval)) + } } close(stream) diff --git a/api/cloudcontroller/ccv3/job_test.go b/api/cloudcontroller/ccv3/job_test.go index 262d938aaf3..abd57bfb1e4 100644 --- a/api/cloudcontroller/ccv3/job_test.go +++ b/api/cloudcontroller/ccv3/job_test.go @@ -188,16 +188,39 @@ var _ = Describe("Job", func() { warnings Warnings executeErr error - startTime time.Time - jobPollingTimeout time.Duration fakeClock *ccv3fakes.FakeClock + pollingSleeps []time.Duration + fakeNow time.Time + fakeGETDuration time.Duration ) + useFakeClock := func() { + fakeNow = time.Time{} + fakeGETDuration = 0 + pollingSleeps = nil + fakeClock.NowCalls(func() time.Time { return fakeNow }) + fakeClock.SleepCalls(func(duration time.Duration) { + pollingSleeps = append(pollingSleeps, duration) + fakeNow = fakeNow.Add(duration) + }) + } + + setInterval := func(interval time.Duration) { + client, fakeClock = NewTestClient(Config{ + JobPollingTimeout: time.Minute, + JobPollingInterval: interval, + }) + useFakeClock() + } + appendHandler := func(state string, warnings Warnings) { server.AppendHandlers( CombineHandlers( VerifyRequest(http.MethodGet, "/some-job-location"), + http.HandlerFunc(func(http.ResponseWriter, *http.Request) { + fakeNow = fakeNow.Add(fakeGETDuration) + }), RespondWith(http.StatusAccepted, fmt.Sprintf(` { "guid": "job-guid", @@ -255,6 +278,7 @@ var _ = Describe("Job", func() { itFinishesWhenCompleteOrFailed := func() { When("the job starts queued and then finishes successfully", func() { BeforeEach(func() { + setInterval(3 * time.Second) appendHandler("PROCESSING", Warnings{"warning-1"}) appendHandler("PROCESSING", Warnings{"warning-2"}) appendHandler("COMPLETE", Warnings{"warning-3", "warning-4"}) @@ -321,18 +345,13 @@ var _ = Describe("Job", func() { BeforeEach(func() { jobPollingTimeout = 100 * time.Millisecond client, fakeClock = NewTestClient(Config{ - JobPollingTimeout: jobPollingTimeout, + JobPollingTimeout: jobPollingTimeout, + JobPollingInterval: 60 * time.Millisecond, }) - - clockTime := time.Now() - fakeClock.NowReturnsOnCall(0, clockTime) - fakeClock.NowReturnsOnCall(1, clockTime) - fakeClock.NowReturnsOnCall(2, clockTime.Add(60*time.Millisecond)) - fakeClock.NowReturnsOnCall(3, clockTime.Add(60*time.Millisecond*2)) + useFakeClock() appendHandler("PROCESSING", Warnings{"warning-1"}) appendHandler("PROCESSING", Warnings{"warning-2", "warning-3"}) - appendHandler("FINISHED", Warnings{"warning-4"}) }) It("raises a JobTimeoutError", func() { @@ -343,42 +362,153 @@ var _ = Describe("Job", func() { Expect(warnings).To(ConsistOf("warning-1", "warning-2", "warning-3")) }) - // Fuzzy test to ensure that the overall function time isn't [far] - // greater than the OverallPollingTimeout. Since this is partially - // dependent on the speed of the system, the expectation is that the - // function *should* never exceed three times the timeout. - It("does not run [too much] longer than the timeout", func() { - endTime := time.Now() - Expect(executeErr).To(HaveOccurred()) - - // If the jobPollingTimeout is less than the PollingInterval, - // then the margin may be too small, we should not allow the - // jobPollingTimeout to be set to less than the PollingInterval - Expect(endTime).To(BeTemporally("~", startTime, 3*jobPollingTimeout)) + It("checks the timeout before another GET", func() { + Expect(server.ReceivedRequests()).To(HaveLen(2)) + Expect(pollingSleeps).To(Equal([]time.Duration{60 * time.Millisecond, 60 * time.Millisecond})) }) }) }) } BeforeEach(func() { - client, _ = NewTestClient(Config{JobPollingTimeout: time.Minute}) + setInterval(3 * time.Second) jobLocation = JobURL(fmt.Sprintf("%s/some-job-location", server.URL())) }) Describe("PollJob", func() { JustBeforeEach(func() { - startTime = time.Now() warnings, executeErr = client.PollJob(jobLocation) }) itSkipsEmptyURLs() itFinishesWhenCompleteOrFailed() itRespectsTimeouts() + + Context("polling delays", func() { + When("using the default interval", func() { + BeforeEach(func() { + setInterval(3 * time.Second) + for range 6 { + appendHandler("PROCESSING", nil) + } + appendHandler("COMPLETE", nil) + }) + + It("starts at 250ms, doubles to the 3s interval, then stays there", func() { + Expect(executeErr).NotTo(HaveOccurred()) + Expect(server.ReceivedRequests()).To(HaveLen(7)) + Expect(pollingSleeps).To(Equal([]time.Duration{ + 250 * time.Millisecond, 500 * time.Millisecond, time.Second, + 1250 * time.Millisecond, 3 * time.Second, 3 * time.Second, + })) + }) + }) + + When("each GET takes 100ms", func() { + BeforeEach(func() { + setInterval(3 * time.Second) + fakeGETDuration = 100 * time.Millisecond + for range 5 { + appendHandler("PROCESSING", nil) + } + appendHandler("COMPLETE", nil) + }) + + It("sleeps 150ms, 400ms, 900ms, 1.15s, and 2.9s to keep checks on the grid", func() { + Expect(executeErr).NotTo(HaveOccurred()) + Expect(pollingSleeps).To(Equal([]time.Duration{ + 150 * time.Millisecond, 400 * time.Millisecond, 900 * time.Millisecond, + 1150 * time.Millisecond, 2900 * time.Millisecond, + })) + }) + }) + + When("the interval is 1s", func() { + BeforeEach(func() { + setInterval(time.Second) + for range 4 { + appendHandler("PROCESSING", nil) + } + appendHandler("COMPLETE", nil) + }) + + It("caps the delay at the interval", func() { + Expect(executeErr).NotTo(HaveOccurred()) + Expect(pollingSleeps).To(Equal([]time.Duration{250 * time.Millisecond, 500 * time.Millisecond, 250 * time.Millisecond, time.Second})) + }) + }) + + When("the interval is below the initial delay", func() { + BeforeEach(func() { + setInterval(100 * time.Millisecond) + appendHandler("PROCESSING", nil) + appendHandler("PROCESSING", nil) + appendHandler("COMPLETE", nil) + }) + + It("uses the interval for every delay", func() { + Expect(executeErr).NotTo(HaveOccurred()) + Expect(pollingSleeps).To(Equal([]time.Duration{100 * time.Millisecond, 100 * time.Millisecond})) + }) + }) + + When("the job completes on the first GET", func() { + BeforeEach(func() { + setInterval(3 * time.Second) + appendHandler("COMPLETE", nil) + }) + + It("does not sleep", func() { + Expect(executeErr).NotTo(HaveOccurred()) + Expect(server.ReceivedRequests()).To(HaveLen(1)) + Expect(pollingSleeps).To(BeEmpty()) + }) + }) + + When("PollJob is called a second time on the same client", func() { + BeforeEach(func() { + setInterval(3 * time.Second) + appendHandler("PROCESSING", nil) + appendHandler("COMPLETE", nil) + }) + + It("starts the second call at 250ms", func() { + appendHandler("PROCESSING", nil) + appendHandler("COMPLETE", nil) + Expect(executeErr).NotTo(HaveOccurred()) + _, secondErr := client.PollJob(jobLocation) + Expect(secondErr).NotTo(HaveOccurred()) + Expect(pollingSleeps).To(Equal([]time.Duration{ + 250 * time.Millisecond, 250 * time.Millisecond, + })) + }) + }) + }) }) + DescribeTable("PollJob accounts for elapsed GET time and the next polling grid point", + func(elapsed, interval, expected time.Duration) { + setInterval(interval) + fakeGETDuration = elapsed + appendHandler("PROCESSING", nil) + appendHandler("COMPLETE", nil) + + _, executeErr = client.PollJob(jobLocation) + Expect(executeErr).NotTo(HaveOccurred()) + Expect(pollingSleeps).To(Equal([]time.Duration{expected})) + }, + + Entry("when elapsed time is 1s and interval is 3s", time.Second, 3*time.Second, 750*time.Millisecond), + Entry("when elapsed time is 8s and interval is 3s", 8*time.Second, 3*time.Second, time.Second), + Entry("when elapsed time is 1.75s and interval is 10s", 1750*time.Millisecond, 10*time.Second, 2*time.Second), + Entry("when elapsed time is exactly on the 3s grid point", 3*time.Second, 3*time.Second, 3*time.Second), + Entry("when elapsed time and interval are zero", time.Duration(0), time.Duration(0), time.Duration(0)), + Entry("when the interval is negative", time.Duration(0), -time.Second, -time.Second), + Entry("when the interval is below the initial delay", time.Duration(0), 100*time.Millisecond, 100*time.Millisecond), + ) + Describe("PollJobForState", func() { JustBeforeEach(func() { - startTime = time.Now() warnings, executeErr = client.PollJobForState(jobLocation, constant.JobPolling) }) @@ -404,7 +534,6 @@ var _ = Describe("Job", func() { var stream chan PollJobEvent JustBeforeEach(func() { - startTime = time.Now() stream = client.PollJobToEventStream(jobLocation) }) @@ -421,12 +550,17 @@ var _ = Describe("Job", func() { When("the job starts queued and then finishes successfully", func() { BeforeEach(func() { + setInterval(3 * time.Second) appendHandler("PROCESSING", Warnings{"warning-1"}) appendHandler("PROCESSING", Warnings{"warning-2"}) + appendHandler("PROCESSING", nil) + appendHandler("PROCESSING", nil) + appendHandler("PROCESSING", nil) + appendHandler("PROCESSING", nil) appendHandler("COMPLETE", Warnings{"warning-3", "warning-4"}) }) - It("should poll until completion", func() { + It("closes the stream after the terminal event without sleeping", func() { Eventually(stream).Should(Receive(Equal(PollJobEvent{ State: constant.JobProcessing, Err: nil, @@ -439,6 +573,9 @@ var _ = Describe("Job", func() { Warnings: Warnings{"warning-2"}, JobGUID: "job-guid", }))) + for range 4 { + Eventually(stream).Should(Receive()) + } Eventually(stream).Should(Receive(Equal(PollJobEvent{ State: constant.JobComplete, Err: nil, @@ -446,6 +583,33 @@ var _ = Describe("Job", func() { JobGUID: "job-guid", }))) Eventually(stream).Should(BeClosed()) + Expect(pollingSleeps).To(Equal([]time.Duration{ + 250 * time.Millisecond, 500 * time.Millisecond, time.Second, + 1250 * time.Millisecond, 3 * time.Second, 3 * time.Second, + })) + }) + }) + + When("a GET takes longer than the next scheduled polling point", func() { + BeforeEach(func() { + setInterval(3 * time.Second) + fakeGETDuration = time.Second + appendHandler("PROCESSING", nil) + appendHandler("COMPLETE", nil) + }) + + It("sleeps until the next polling grid point", func() { + Eventually(stream).Should(Receive(Equal(PollJobEvent{ + State: constant.JobProcessing, + JobGUID: "job-guid", + }))) + Eventually(stream).Should(Receive(Equal(PollJobEvent{ + State: constant.JobComplete, + JobGUID: "job-guid", + }))) + Eventually(stream).Should(BeClosed()) + Expect(server.ReceivedRequests()).To(HaveLen(2)) + Expect(pollingSleeps).To(Equal([]time.Duration{750 * time.Millisecond})) }) }) @@ -484,14 +648,11 @@ var _ = Describe("Job", func() { BeforeEach(func() { jobPollingTimeout = 100 * time.Millisecond client, fakeClock = NewTestClient(Config{ - JobPollingTimeout: jobPollingTimeout, + JobPollingTimeout: jobPollingTimeout, + JobPollingInterval: 60 * time.Millisecond, }) - clockTime := time.Now() - fakeClock.NowReturnsOnCall(0, clockTime) - fakeClock.NowReturnsOnCall(1, clockTime) - fakeClock.NowReturnsOnCall(2, clockTime.Add(60*time.Millisecond)) - fakeClock.NowReturnsOnCall(3, clockTime.Add(60*time.Millisecond*2)) + useFakeClock() appendHandler("PROCESSING", Warnings{"warning-1"}) appendHandler("PROCESSING", Warnings{"warning-2", "warning-3"}) @@ -511,18 +672,13 @@ var _ = Describe("Job", func() { Eventually(stream).Should(BeClosed()) }) - // Fuzzy test to ensure that the overall function time isn't [far] - // greater than the OverallPollingTimeout. Since this is partially - // dependent on the speed of the system, the expectation is that the - // function *should* never exceed three times the timeout. - It("does not run [too much] longer than the timeout", func() { + It("checks timeout after GET and sleeps only between events", func() { + for range 3 { + Eventually(stream).Should(Receive()) + } Eventually(stream).Should(BeClosed()) - endTime := time.Now() - - // If the jobPollingTimeout is less than the PollingInterval, - // then the margin may be too small, we should not allow the - // jobPollingTimeout to be set to less than the PollingInterval - Expect(endTime).To(BeTemporally("~", startTime, 3*jobPollingTimeout)) + Expect(server.ReceivedRequests()).To(HaveLen(3)) + Expect(pollingSleeps).To(Equal([]time.Duration{60 * time.Millisecond, 60 * time.Millisecond})) }) }) })