diff --git a/internal/dev_server/adapters/api.go b/internal/dev_server/adapters/api.go index c968b3ac..344bbb44 100644 --- a/internal/dev_server/adapters/api.go +++ b/internal/dev_server/adapters/api.go @@ -47,7 +47,7 @@ func (a apiClientApi) GetSdkKey(ctx context.Context, projectKey, environmentKey func (a apiClientApi) GetAllFlags(ctx context.Context, projectKey string) ([]ldapi.FeatureFlag, error) { log.Printf("Fetching all flags for project '%s'", projectKey) - flags, err := a.getFlags(ctx, projectKey, nil) + flags, err := a.getFlags(ctx, projectKey) if err != nil { err = errors.Wrap(err, "unable to get all flags from LD API") } @@ -63,8 +63,8 @@ func (a apiClientApi) GetProjectEnvironments(ctx context.Context, projectKey str return environments, err } -func (a apiClientApi) getFlags(ctx context.Context, projectKey string, href *string) ([]ldapi.FeatureFlag, error) { - return internal.GetPaginatedItems(ctx, projectKey, href, func(ctx context.Context, projectKey string, limit, offset *int64) (flags *ldapi.FeatureFlags, err error) { +func (a apiClientApi) getFlags(ctx context.Context, projectKey string) ([]ldapi.FeatureFlag, error) { + return internal.GetPaginatedItems(ctx, projectKey, func(ctx context.Context, projectKey string, limit, offset *int64) (flags *ldapi.FeatureFlags, err error) { // loop until we do not get rate limited query := a.apiClient.FeatureFlagsApi.GetFeatureFlags(ctx, projectKey).Limit(100) query = query.Filter("purpose:all+!(holdout)") diff --git a/internal/dev_server/adapters/internal/api_util.go b/internal/dev_server/adapters/internal/api_util.go index 8a2b3d33..9b38afd6 100644 --- a/internal/dev_server/adapters/internal/api_util.go +++ b/internal/dev_server/adapters/internal/api_util.go @@ -3,76 +3,92 @@ package internal import ( "context" "log" + "math/rand/v2" "net/http" - "net/url" "strconv" + "sync" "time" "github.com/launchdarkly/api-client-go/v14" "github.com/pkg/errors" ) +// maxConcurrentPageFetches bounds how many pages we fetch at once once the total +// item count is known. Kept modest to stay well clear of the API's rate limiter; +// Retry429s still handles any 429s that slip through. +const maxConcurrentPageFetches = 6 + +// GetPaginatedItems fetches all pages of a paginated list endpoint. It fetches +// page 0 first to learn the total item count, then fetches the remaining pages +// concurrently (bounded by maxConcurrentPageFetches) instead of following the +// `next` link serially, since every offset is already known once the total +// count and page size are known. func GetPaginatedItems[T any, R interface { GetItems() []T GetLinks() map[string]ldapi.Link -}](ctx context.Context, projectKey string, href *string, fetchFunc func(context.Context, string, *int64, *int64) (R, error)) ([]T, error) { - var result R - var err error - - if href == nil { - result, err = fetchFunc(ctx, projectKey, nil, nil) - if err != nil { - return nil, err - } - } else { - limit, offset, err := parseHref(*href) - if err != nil { - return nil, errors.Wrapf(err, "unable to parse href for next link: %s", *href) - } - result, err = fetchFunc(ctx, projectKey, &limit, &offset) - if err != nil { - return nil, err - } + GetTotalCount() int32 +}](ctx context.Context, projectKey string, fetchFunc func(context.Context, string, *int64, *int64) (R, error)) ([]T, error) { + first, err := fetchFunc(ctx, projectKey, nil, nil) + if err != nil { + return nil, err } - items := result.GetItems() - - if links := result.GetLinks(); links != nil { - if next, ok := links["next"]; ok && next.Href != nil { - newItems, err := GetPaginatedItems(ctx, projectKey, next.Href, fetchFunc) - if err != nil { - return nil, err - } - items = append(items, newItems...) - } + firstItems := first.GetItems() + limit := int64(len(firstItems)) + total := int64(first.GetTotalCount()) + if limit == 0 || total <= limit { + return firstItems, nil } - return items, nil -} + numPages := int((total + limit - 1) / limit) + pages := make([][]T, numPages) + pages[0] = firstItems -func parseHref(href string) (limit, offset int64, err error) { - parsedUrl, err := url.Parse(href) - if err != nil { - return - } - l, err := strconv.Atoi(parsedUrl.Query().Get("limit")) - if err != nil { - return + sem := make(chan struct{}, maxConcurrentPageFetches) + var wg sync.WaitGroup + errs := make([]error, numPages) + + for page := 1; page < numPages; page++ { + page := page + offset := int64(page) * limit + wg.Add(1) + sem <- struct{}{} + go func() { + defer wg.Done() + defer func() { <-sem }() + result, fetchErr := fetchFunc(ctx, projectKey, &limit, &offset) + if fetchErr != nil { + errs[page] = fetchErr + return + } + pages[page] = result.GetItems() + }() } - o, err := strconv.Atoi(parsedUrl.Query().Get("offset")) - if err != nil { - return + wg.Wait() + + for _, pageErr := range errs { + if pageErr != nil { + return nil, pageErr + } } - limit = int64(l) - offset = int64(o) - return + items := make([]T, 0, total) + for _, p := range pages { + items = append(items, p...) + } + return items, nil } //go:generate go run go.uber.org/mock/mockgen -destination mocks.go -package internal . MockableTime type MockableTime interface { Sleep(duration time.Duration) Now() time.Time + // Jitter returns a random duration in [0, max). Used to decorrelate + // retries after a 429: with concurrent page fetches, several requests + // can get rate-limited at once and would otherwise all compute the same + // X-Ratelimit-Reset and wake up to retry at the exact same instant, + // recreating the burst that got them limited in the first place. + Jitter(max time.Duration) time.Duration } type realTime struct{} @@ -85,8 +101,20 @@ func (realTime) Now() time.Time { return time.Now() } +func (realTime) Jitter(max time.Duration) time.Duration { + if max <= 0 { + return 0 + } + return rand.N(max) +} + var timeImpl MockableTime = realTime{} +// maxRetryJitter bounds the random extra delay added on top of the API's +// requested X-Ratelimit-Reset wait, so concurrent retries spread out instead +// of waking up in lockstep. +const maxRetryJitter = 250 * time.Millisecond + func Retry429s[T any](requester func() (T, *http.Response, error)) (result T, err error) { for { var res *http.Response @@ -98,9 +126,9 @@ func Retry429s[T any](requester func() (T, *http.Response, error)) (result T, er err = errors.Wrapf(err, `unable to retry rate limited request: X-RateLimit-Reset: "%s" was not parsable`, resetUnixMillisString) return } - sleep := resetUnixMillis - timeImpl.Now().UnixMilli() - log.Printf("Got 429 in API response. Retrying in %d milliseconds.", sleep) - timeImpl.Sleep(time.Duration(sleep) * time.Millisecond) + sleep := time.Duration(resetUnixMillis-timeImpl.Now().UnixMilli())*time.Millisecond + timeImpl.Jitter(maxRetryJitter) + log.Printf("Got 429 in API response. Retrying in %s.", sleep) + timeImpl.Sleep(sleep) } else { return } diff --git a/internal/dev_server/adapters/internal/api_util_test.go b/internal/dev_server/adapters/internal/api_util_test.go index 28d388d0..a4eeb873 100644 --- a/internal/dev_server/adapters/internal/api_util_test.go +++ b/internal/dev_server/adapters/internal/api_util_test.go @@ -3,6 +3,8 @@ package internal import ( "context" "net/http" + "strconv" + "sync" "testing" "time" @@ -16,8 +18,9 @@ type testItem struct { } type testResult struct { - items []testItem - links map[string]ldapi.Link + items []testItem + links map[string]ldapi.Link + totalCount int32 } func (r testResult) GetItems() []testItem { @@ -28,122 +31,95 @@ func (r testResult) GetLinks() map[string]ldapi.Link { return r.links } +func (r testResult) GetTotalCount() int32 { + return r.totalCount +} + func TestGetPaginatedItems(t *testing.T) { ctx := context.Background() projectKey := "test-project" - testCases := []struct { - name string - fetchResponses []testResult - expectedItems []testItem - expectedError bool - }{ - { - name: "Single page", - fetchResponses: []testResult{ - { - items: []testItem{{ID: "1"}, {ID: "2"}}, - links: map[string]ldapi.Link{}, - }, - }, - expectedItems: []testItem{{ID: "1"}, {ID: "2"}}, - }, - { - name: "Multiple pages", - fetchResponses: []testResult{ - { - items: []testItem{{ID: "1"}, {ID: "2"}}, - links: map[string]ldapi.Link{ - "next": {Href: strPtr("http://example.com?limit=2&offset=2")}, - }, - }, - { - items: []testItem{{ID: "3"}, {ID: "4"}}, - links: map[string]ldapi.Link{}, - }, - }, - expectedItems: []testItem{{ID: "1"}, {ID: "2"}, {ID: "3"}, {ID: "4"}}, - }, - { - name: "Error on second page", - fetchResponses: []testResult{ - { - items: []testItem{{ID: "1"}, {ID: "2"}}, - links: map[string]ldapi.Link{ - "next": {Href: strPtr("http://example.com?limit=2&offset=2")}, - }, - }, - }, - expectedError: true, - }, - { - name: "Empty response", - fetchResponses: []testResult{ - { - items: []testItem{}, - links: map[string]ldapi.Link{}, - }, - }, - expectedItems: []testItem{}, - }, - { - name: "Multiple pages with varying item counts", - fetchResponses: []testResult{ - { - items: []testItem{{ID: "1"}, {ID: "2"}, {ID: "3"}}, - links: map[string]ldapi.Link{ - "next": {Href: strPtr("http://example.com?limit=3&offset=3")}, - }, - }, - { - items: []testItem{{ID: "4"}, {ID: "5"}}, - links: map[string]ldapi.Link{ - "next": {Href: strPtr("http://example.com?limit=3&offset=5")}, - }, - }, - { - items: []testItem{{ID: "6"}}, - links: map[string]ldapi.Link{}, - }, - }, - expectedItems: []testItem{{ID: "1"}, {ID: "2"}, {ID: "3"}, {ID: "4"}, {ID: "5"}, {ID: "6"}}, - }, - { - name: "Invalid next link", - fetchResponses: []testResult{ - { - items: []testItem{{ID: "1"}, {ID: "2"}}, - links: map[string]ldapi.Link{ - "next": {Href: strPtr("invalid-url")}, - }, - }, - }, - expectedError: true, - }, - } - - for _, tc := range testCases { - t.Run(tc.name, func(t *testing.T) { - callCount := 0 - fetchFunc := func(ctx context.Context, projectKey string, limit, offset *int64) (testResult, error) { - if callCount >= len(tc.fetchResponses) { - return testResult{}, assert.AnError - } - result := tc.fetchResponses[callCount] - callCount++ - return result, nil + t.Run("single page", func(t *testing.T) { + fetchFunc := func(ctx context.Context, projectKey string, limit, offset *int64) (testResult, error) { + assert.Nil(t, limit) + assert.Nil(t, offset) + return testResult{items: []testItem{{ID: "1"}, {ID: "2"}}, totalCount: 2}, nil + } + + items, err := GetPaginatedItems(ctx, projectKey, fetchFunc) + assert.NoError(t, err) + assert.Equal(t, []testItem{{ID: "1"}, {ID: "2"}}, items) + }) + + t.Run("multiple pages fetched by offset, order preserved regardless of arrival order", func(t *testing.T) { + pages := map[int64][]testItem{ + 0: {{ID: "1"}, {ID: "2"}}, + 2: {{ID: "3"}, {ID: "4"}}, + 4: {{ID: "5"}}, + } + + fetchFunc := func(ctx context.Context, projectKey string, limit, offset *int64) (testResult, error) { + if offset == nil { + return testResult{items: pages[0], totalCount: 5}, nil } + return testResult{items: pages[*offset]}, nil + } - items, err := GetPaginatedItems(ctx, projectKey, nil, fetchFunc) + items, err := GetPaginatedItems(ctx, projectKey, fetchFunc) + assert.NoError(t, err) + assert.Equal(t, []testItem{{ID: "1"}, {ID: "2"}, {ID: "3"}, {ID: "4"}, {ID: "5"}}, items) + }) - if tc.expectedError { - assert.Error(t, err) - } else { - assert.NoError(t, err) - assert.Equal(t, tc.expectedItems, items) + t.Run("error on any page propagates", func(t *testing.T) { + fetchFunc := func(ctx context.Context, projectKey string, limit, offset *int64) (testResult, error) { + if offset == nil { + return testResult{items: []testItem{{ID: "1"}, {ID: "2"}}, totalCount: 4}, nil } - }) - } + return testResult{}, assert.AnError + } + + _, err := GetPaginatedItems(ctx, projectKey, fetchFunc) + assert.Error(t, err) + }) + + t.Run("empty response", func(t *testing.T) { + fetchFunc := func(ctx context.Context, projectKey string, limit, offset *int64) (testResult, error) { + return testResult{items: []testItem{}, totalCount: 0}, nil + } + + items, err := GetPaginatedItems(ctx, projectKey, fetchFunc) + assert.NoError(t, err) + assert.Equal(t, []testItem{}, items) + }) + + t.Run("remaining pages are fetched concurrently, not one at a time", func(t *testing.T) { + var mu sync.Mutex + inFlight := 0 + maxInFlight := 0 + + fetchFunc := func(ctx context.Context, projectKey string, limit, offset *int64) (testResult, error) { + if offset == nil { + return testResult{items: []testItem{{ID: "0"}}, totalCount: 4}, nil + } + mu.Lock() + inFlight++ + if inFlight > maxInFlight { + maxInFlight = inFlight + } + mu.Unlock() + + time.Sleep(20 * time.Millisecond) + + mu.Lock() + inFlight-- + mu.Unlock() + return testResult{items: []testItem{{ID: "x"}}}, nil + } + + _, err := GetPaginatedItems(ctx, projectKey, fetchFunc) + assert.NoError(t, err) + assert.Greater(t, maxInFlight, 1, "expected more than one page in flight at once") + }) } func TestRetry429s(t *testing.T) { @@ -158,14 +134,15 @@ func TestRetry429s(t *testing.T) { assert.Equal(t, 1, called) }) - t.Run("it should retry when a 429 is received", func(t *testing.T) { + t.Run("it should retry when a 429 is received, adding jitter on top of the reset wait", func(t *testing.T) { ctrl := gomock.NewController(t) timeMock := NewMockMockableTime(ctrl) defer func() { ctrl.Finish() }() timeImpl = timeMock defer func() { timeImpl = realTime{} }() timeMock.EXPECT().Now().Return(time.UnixMilli(0)) - timeMock.EXPECT().Sleep(time.Duration(1000) * time.Millisecond) + timeMock.EXPECT().Jitter(maxRetryJitter).Return(50 * time.Millisecond) + timeMock.EXPECT().Sleep(1050 * time.Millisecond) called := 0 res, err := Retry429s(func() (string, *http.Response, error) { @@ -182,8 +159,45 @@ func TestRetry429s(t *testing.T) { assert.NoError(t, err) assert.Equal(t, 2, called) }) -} -func strPtr(s string) *string { - return &s + t.Run("concurrent retries get decorrelated jitter instead of waking up in lockstep", func(t *testing.T) { + // Simulates several concurrent page fetches all getting rate-limited + // with the same X-Ratelimit-Reset at once. With realTime's actual + // Jitter (not mocked here), their computed sleep durations should + // differ instead of being identical, so they don't all retry at the + // exact same instant and recreate the burst that got them limited. + reset := time.Now().Add(20 * time.Millisecond).UnixMilli() + resetStr := strconv.FormatInt(reset, 10) + + const n = 8 + sleeps := make([]time.Duration, n) + var wg sync.WaitGroup + for i := 0; i < n; i++ { + i := i + wg.Add(1) + go func() { + defer wg.Done() + called := 0 + start := time.Now() + _, err := Retry429s(func() (string, *http.Response, error) { + called++ + if called > 1 { + return "ok", &http.Response{StatusCode: 200}, nil + } + header := make(http.Header) + header.Set("X-Ratelimit-Reset", resetStr) + return "", &http.Response{StatusCode: 429, Header: header}, nil + }) + assert.NoError(t, err) + sleeps[i] = time.Since(start) + }() + } + wg.Wait() + + distinct := map[time.Duration]bool{} + for _, s := range sleeps { + distinct[s] = true + } + assert.Greater(t, len(distinct), 1, "expected jittered sleep durations to differ across concurrent retries") + }) } diff --git a/internal/dev_server/adapters/internal/mocks.go b/internal/dev_server/adapters/internal/mocks.go index 6a4dd2fa..99af1a7f 100644 --- a/internal/dev_server/adapters/internal/mocks.go +++ b/internal/dev_server/adapters/internal/mocks.go @@ -40,6 +40,20 @@ func (m *MockMockableTime) EXPECT() *MockMockableTimeMockRecorder { return m.recorder } +// Jitter mocks base method. +func (m *MockMockableTime) Jitter(max time.Duration) time.Duration { + m.ctrl.T.Helper() + ret := m.ctrl.Call(m, "Jitter", max) + ret0, _ := ret[0].(time.Duration) + return ret0 +} + +// Jitter indicates an expected call of Jitter. +func (mr *MockMockableTimeMockRecorder) Jitter(max any) *gomock.Call { + mr.mock.ctrl.T.Helper() + return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "Jitter", reflect.TypeOf((*MockMockableTime)(nil).Jitter), max) +} + // Now mocks base method. func (m *MockMockableTime) Now() time.Time { m.ctrl.T.Helper()