perf: Speed up dev server startup (two flavors: parallelism and streaming)

cmd/dev_server/flags.go

@@ -4,4 +4,9 @@ const (
 	ContextFlag           = "context"
 	OverrideFlag          = "override"
 	SourceEnvironmentFlag = "source"
+
+	StreamFlagStartupFlag        = "stream-flag-startup"
+	StreamFlagStartupDescription = "Load flag values from the streaming connection at startup and resolve variation " +
+		"display names from REST in the background. Speeds up startup on large projects (the health check passes in " +
+		"~1s) at the cost of variation names appearing in the UI a few seconds later."
 )

cmd/dev_server/start_server.go

@@ -46,6 +46,9 @@ func NewStartServerCmd(client dev_server.Client) *cobra.Command {
 	cmd.Flags().Bool(cliflags.SyncOnceFlag, false, cliflags.SyncOnceFlagDescription)
 	_ = viper.BindPFlag(cliflags.SyncOnceFlag, cmd.Flags().Lookup(cliflags.SyncOnceFlag))
 
+	cmd.Flags().Bool(StreamFlagStartupFlag, false, StreamFlagStartupDescription)
+	_ = viper.BindPFlag(StreamFlagStartupFlag, cmd.Flags().Lookup(StreamFlagStartupFlag))
+
 	return cmd
 }
 
@@ -91,6 +94,7 @@ func startServer(client dev_server.Client) func(*cobra.Command, []string) error
 			Port:                   viper.GetString(cliflags.PortFlag),
 			CorsEnabled:            viper.GetBool(cliflags.CorsEnabledFlag),
 			CorsOrigin:             viper.GetString(cliflags.CorsOriginFlag),
+			StreamFlagStartup:      viper.GetBool(StreamFlagStartupFlag),
 			InitialProjectSettings: initialSetting,
 		}
 

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,23 +63,30 @@ 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) {
-		// loop until we do not get rate limited
-		query := a.apiClient.FeatureFlagsApi.GetFeatureFlags(ctx, projectKey).Limit(100)
-		query = query.Filter("purpose:all+!(holdout)")
-
-		if limit != nil {
-			query = query.Limit(*limit)
-		}
+const (
+	flagsPageSize    = 100
+	flagsConcurrency = 6
+)
 
-		if offset != nil {
-			query = query.Offset(*offset)
-		}
-		return internal.Retry429s(query.Execute)
+// getFlags pages the flags list concurrently (see internal.FetchPagesConcurrently).
+func (a apiClientApi) getFlags(ctx context.Context, projectKey string) ([]ldapi.FeatureFlag, error) {
+	return internal.FetchPagesConcurrently(flagsPageSize, flagsConcurrency, func(offset int64) ([]ldapi.FeatureFlag, error) {
+		return a.getFlagsPage(ctx, projectKey, offset)
 	})
 }
 
+func (a apiClientApi) getFlagsPage(ctx context.Context, projectKey string, offset int64) ([]ldapi.FeatureFlag, error) {
+	query := a.apiClient.FeatureFlagsApi.GetFeatureFlags(ctx, projectKey).
+		Filter("purpose:all+!(holdout)").
+		Limit(flagsPageSize).
+		Offset(offset)
+	flags, err := internal.Retry429s(query.Execute)
+	if err != nil {
+		return nil, err
+	}
+	return flags.Items, nil
+}
+
 func (a apiClientApi) getEnvironments(ctx context.Context, projectKey string, href *string, query string, limit *int) ([]ldapi.Environment, error) {
 	request := a.apiClient.EnvironmentsApi.GetEnvironmentsByProject(ctx, projectKey)
 

internal/dev_server/adapters/internal/api_util.go

@@ -1,72 +1,50 @@
 package internal
 
 import (
-	"context"
 	"log"
 	"net/http"
-	"net/url"
 	"strconv"
+	"sync"
 	"time"
 
-	"github.com/launchdarkly/api-client-go/v14"
 	"github.com/pkg/errors"
 )
 
-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
-		}
+// FetchPagesConcurrently returns every item across an offset-paginated list, fetching pages in bounded concurrent batches until a short page ends the list. It never trusts a reported total count (0 when absent would truncate); a short page is the only end signal.
+func FetchPagesConcurrently[T any](pageSize, concurrency int, fetch func(offset int64) ([]T, error)) ([]T, error) {
+	first, err := fetch(0)
+	if err != nil {
+		return nil, err
+	}
+	all := first
+	if len(first) < pageSize {
+		return all, nil
 	}
 
-	items := result.GetItems()
+	for offset := int64(pageSize); ; offset += int64(concurrency) * int64(pageSize) {
+		batch := make([][]T, concurrency)
+		errs := make([]error, concurrency)
+		var wg sync.WaitGroup
+		for i := 0; i < concurrency; i++ {
+			wg.Add(1)
+			go func(i int) {
+				defer wg.Done()
+				batch[i], errs[i] = fetch(offset + int64(i)*int64(pageSize))
+			}(i)
+		}
+		wg.Wait()
 
-	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
+		// Stop at the first short/empty page; pages after it are speculative probes past the end, so ignore their errors.
+		for i := 0; i < concurrency; i++ {
+			if errs[i] != nil {
+				return nil, errs[i]
+			}
+			all = append(all, batch[i]...)
+			if len(batch[i]) < pageSize {
+				return all, nil
 			}
-			items = append(items, newItems...)
 		}
 	}
-
-	return items, nil
-}
-
-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
-	}
-	o, err := strconv.Atoi(parsedUrl.Query().Get("offset"))
-	if err != nil {
-		return
-	}
-
-	limit = int64(l)
-	offset = int64(o)
-	return
 }
 
 //go:generate go run go.uber.org/mock/mockgen -destination mocks.go -package internal . MockableTime
@@ -91,6 +69,10 @@ func Retry429s[T any](requester func() (T, *http.Response, error)) (result T, er
 	for {
 		var res *http.Response
 		result, res, err = requester()
+		// Only bail on a nil response (transport failure); a 429 comes back as a non-nil error with a non-nil response, so never bail on err alone or the retry is skipped.
+		if res == nil {
+			return
+		}
 		if res.StatusCode == 429 {
 			resetUnixMillisString := res.Header.Get("X-Ratelimit-Reset")
 			resetUnixMillis, strconvErr := strconv.ParseInt(resetUnixMillisString, 10, 64)