fix: Optimize creation of changesets from collections (#335)

Previously, we worked to cache the changset -> collections process as it
is used in many different places. There are multiple places where a
changeset is created from a collection. With this change, we take
advantage of this knowledge, and pre-cache the collection that created
the changeset, preventing us from doing the conversion at all.

Author: keelerm84

internal/datasourcev2/fdv1_polling_data_source.go

@@ -2,11 +2,9 @@ package datasourcev2
 
 import (
 	"context"
-	"encoding/json"
 	"net/http"
 
 	"github.com/launchdarkly/go-sdk-common/v3/ldlog"
-	"github.com/launchdarkly/go-server-sdk/v7/internal/datakinds"
 	"github.com/launchdarkly/go-server-sdk/v7/internal/datasource"
 	"github.com/launchdarkly/go-server-sdk/v7/subsystems"
 )
@@ -33,38 +31,22 @@ func (r *fdv1ToFDv2Requester) Request(
 		reason = "cached"
 	}
 
-	changeSetBuilder := subsystems.NewChangeSetBuilder()
-	changeSetBuilder.Start(subsystems.ServerIntent{
+	intent := subsystems.ServerIntent{
 		Payload: subsystems.Payload{
 			ID:     "arbitrary-id",
 			Target: 0,
 			Code:   code,
 			Reason: reason,
 		},
-	})
-
-	for _, item := range data {
-		kind := subsystems.FlagKind
-		if item.Kind == datakinds.Segments {
-			kind = subsystems.SegmentKind
-		}
-
-		for _, keyedItem := range item.Items {
-			if keyedItem.Item.Item == nil {
-				continue
-			}
-
-			bytes, err := json.Marshal(keyedItem.Item.Item)
-			if err != nil {
-				r.loggers.Warn("Error marshalling v1 item to JSON: %s", err)
-				return nil, headers, err
-			}
-
-			changeSetBuilder.AddPut(kind, keyedItem.Key, keyedItem.Item.Version, bytes)
-		}
 	}
 
-	changeset, err := changeSetBuilder.Finish(subsystems.NewSelector("state", 1))
+	// Data is already in Collection format, use NewChangeSetFromCollections
+	// to avoid the overhead of converting Collections -> Changes -> Collections
+	changeset, err := subsystems.NewChangeSetFromCollections(
+		intent,
+		subsystems.NewSelector("state", 1),
+		data,
+	)
 	return changeset, headers, err
 }
 

ldfiledatav2/file_data_source_impl.go

@@ -20,6 +20,7 @@ import (
 	"github.com/launchdarkly/go-server-sdk/v7/interfaces"
 	"github.com/launchdarkly/go-server-sdk/v7/internal"
 	"github.com/launchdarkly/go-server-sdk/v7/subsystems"
+	"github.com/launchdarkly/go-server-sdk/v7/subsystems/ldstoreimpl"
 	"github.com/launchdarkly/go-server-sdk/v7/subsystems/ldstoretypes"
 
 	"gopkg.in/ghodss/yaml.v1"
@@ -247,12 +248,11 @@ type fileData struct {
 	Segments   *map[string]ldmodel.Segment
 }
 
-func insertData(
+func insertDataIntoCollection(
+	items *[]ldstoretypes.KeyedItemDescriptor,
 	seenKeys map[subsystems.ObjectKind]map[string]bool,
-	builder *subsystems.ChangeSetBuilder,
 	objectKind subsystems.ObjectKind,
 	key string,
-	version int,
 	data ldstoretypes.ItemDescriptor,
 	duplicateKeysHandling DuplicateKeysHandling,
 ) error {
@@ -265,12 +265,10 @@ func insertData(
 		}
 	}
 
-	json, err := json.Marshal(data.Item)
-	if err != nil {
-		return fmt.Errorf("error serializing %s '%s': %w", objectKind, key, err)
-	}
-
-	builder.AddPut(objectKind, key, version, json)
+	*items = append(*items, ldstoretypes.KeyedItemDescriptor{
+		Key:  key,
+		Item: data,
+	})
 	seenKeys[objectKind][key] = true
 
 	return nil
@@ -304,25 +302,29 @@ func mergeFileData(
 	version int,
 	allFileData ...fileData,
 ) (*subsystems.ChangeSet, error) {
-	builder := subsystems.NewChangeSetBuilder()
-	builder.Start(subsystems.ServerIntent{
+	intent := subsystems.ServerIntent{
 		Payload: subsystems.Payload{
 			ID:     "",
 			Target: version,
 			Code:   subsystems.IntentTransferFull,
 			Reason: "payload-missing",
 		},
-	})
+	}
+
+	// Build collections directly instead of using ChangeSetBuilder
+	flagItems := make([]ldstoretypes.KeyedItemDescriptor, 0)
+	segmentItems := make([]ldstoretypes.KeyedItemDescriptor, 0)
 
 	seenKeys := map[subsystems.ObjectKind]map[string]bool{
 		subsystems.FlagKind:    {},
 		subsystems.SegmentKind: {},
 	}
+
 	for _, d := range allFileData {
 		if d.Flags != nil {
 			for key, f := range *d.Flags {
 				data := ldstoretypes.ItemDescriptor{Version: f.Version, Item: &f}
-				err := insertData(seenKeys, builder, subsystems.FlagKind, key, f.Version, data, duplicateKeysHandling)
+				err := insertDataIntoCollection(&flagItems, seenKeys, subsystems.FlagKind, key, data, duplicateKeysHandling)
 				if err != nil {
 					return nil, err
 				}
@@ -332,7 +334,7 @@ func mergeFileData(
 			for key, value := range *d.FlagValues {
 				flag := makeFlagWithValue(key, value)
 				data := ldstoretypes.ItemDescriptor{Version: flag.Version, Item: flag}
-				err := insertData(seenKeys, builder, subsystems.FlagKind, key, flag.Version, data, duplicateKeysHandling)
+				err := insertDataIntoCollection(&flagItems, seenKeys, subsystems.FlagKind, key, data, duplicateKeysHandling)
 				if err != nil {
 					return nil, err
 				}
@@ -341,19 +343,34 @@ func mergeFileData(
 		if d.Segments != nil {
 			for key, s := range *d.Segments {
 				data := ldstoretypes.ItemDescriptor{Version: s.Version, Item: &s}
-				err := insertData(seenKeys, builder, subsystems.SegmentKind, key, s.Version, data, duplicateKeysHandling)
+				err := insertDataIntoCollection(&segmentItems, seenKeys, subsystems.SegmentKind, key, data, duplicateKeysHandling)
 				if err != nil {
 					return nil, err
 				}
 			}
 		}
 	}
 
+	// Build collections
+	collections := make([]ldstoretypes.Collection, 0, 2)
+	if len(flagItems) > 0 {
+		collections = append(collections, ldstoretypes.Collection{
+			Kind:  ldstoreimpl.Features(),
+			Items: flagItems,
+		})
+	}
+	if len(segmentItems) > 0 {
+		collections = append(collections, ldstoretypes.Collection{
+			Kind:  ldstoreimpl.Segments(),
+			Items: segmentItems,
+		})
+	}
+
 	// File data source will not have a selector for now.
 	// NOTE: If we start supporting FDv2 data from file then this statement might change.
 	// When that happens we will construct the selector the same way that we construct it
 	// in the FDv2 polling data source.
-	return builder.Finish(subsystems.NoSelector())
+	return subsystems.NewChangeSetFromCollections(intent, subsystems.NoSelector(), collections)
 }
 
 func makeFlagWithValue(key string, v interface{}) *ldmodel.FeatureFlag {

subsystems/changeset.go

@@ -160,6 +160,86 @@ func NewChangeSetBuilder() *ChangeSetBuilder {
 	return &ChangeSetBuilder{}
 }
 
+// NewChangeSetFromCollections creates a ChangeSet directly from collections,
+// pre-populating the collections cache to avoid redundant conversions.
+//
+// This is useful when data is already in Collection format (e.g., from file data sources
+// or offline mode) and avoids the overhead of converting Collections -> Changes -> Collections.
+//
+// This function is not stable, and not subject to any backwards
+// compatibility guarantees or semantic versioning. It is not suitable for production usage.
+//
+// Do not use it.
+// You have been warned.
+func NewChangeSetFromCollections(
+	intent ServerIntent,
+	selector Selector,
+	collections []ldstoretypes.Collection,
+) (*ChangeSet, error) {
+	// Build Changes from collections for the ChangeSet API
+	changes, err := collectionsToChanges(collections)
+	if err != nil {
+		return nil, err
+	}
+
+	return &ChangeSet{
+		intentCode: intent.Payload.Code,
+		selector:   selector,
+		changes:    changes,
+		collection: collections, // Pre-populate the cache!
+		mu:         &sync.Mutex{},
+	}, nil
+}
+
+// collectionsToChanges converts Collections to Changes.
+// This is the inverse of toStorableItems.
+func collectionsToChanges(collections []ldstoretypes.Collection) ([]Change, error) {
+	var changes []Change
+
+	for _, coll := range collections {
+		// Map from FDv1 DataKind to ObjectKind
+		// Features -> FlagKind, Segments -> SegmentKind
+		kindName := coll.Kind.GetName()
+		var kind ObjectKind
+		switch kindName {
+		case "features":
+			kind = FlagKind
+		case "segments":
+			kind = SegmentKind
+		default:
+			// Unknown kind, skip for forward compatibility
+			continue
+		}
+
+		for _, item := range coll.Items {
+			if item.Item.Item != nil {
+				// This is a Put operation
+				jsonData, err := json.Marshal(item.Item.Item)
+				if err != nil {
+					return nil, err
+				}
+				changes = append(changes, Change{
+					Action:  ChangeTypePut,
+					Kind:    kind,
+					Key:     item.Key,
+					Version: item.Item.Version,
+					Object:  jsonData,
+				})
+			} else {
+				// This is a Delete operation (tombstone)
+				changes = append(changes, Change{
+					Action:  ChangeTypeDelete,
+					Kind:    kind,
+					Key:     item.Key,
+					Version: item.Item.Version,
+				})
+			}
+		}
+	}
+
+	return changes, nil
+}
+
 // NoChanges represents an intent that the current data is up-to-date and doesn't
 // require changes.
 func (c *ChangeSetBuilder) NoChanges() *ChangeSet {

testhelpers/ldtestdatav2/test_data_source.go

@@ -10,6 +10,7 @@ import (
 	"github.com/launchdarkly/go-server-sdk/v7/interfaces"
 	"github.com/launchdarkly/go-server-sdk/v7/internal"
 	"github.com/launchdarkly/go-server-sdk/v7/subsystems"
+	"github.com/launchdarkly/go-server-sdk/v7/subsystems/ldstoreimpl"
 	"github.com/launchdarkly/go-server-sdk/v7/subsystems/ldstoretypes"
 )
 
@@ -205,33 +206,51 @@ func (t *TestDataSynchronizer) makeFullTransferChangeset() (*subsystems.ChangeSe
 	version := t.version
 	t.version++
 
-	builder := subsystems.NewChangeSetBuilder().
-		Start(subsystems.ServerIntent{
-			Payload: subsystems.Payload{
-				ID:     "",
-				Target: version,
-				Code:   subsystems.IntentTransferFull,
-				Reason: "payload-missing",
-			},
-		})
+	intent := subsystems.ServerIntent{
+		Payload: subsystems.Payload{
+			ID:     "",
+			Target: version,
+			Code:   subsystems.IntentTransferFull,
+			Reason: "payload-missing",
+		},
+	}
 
+	// Build collections directly from the current state
+	flagItems := make([]ldstoretypes.KeyedItemDescriptor, 0, len(t.currentFlags))
 	for key, item := range t.currentFlags {
-		json, err := json.Marshal(item.Item)
-		if err != nil {
-			return nil, err
-		}
-		builder.AddPut(subsystems.FlagKind, key, item.Version, json)
+		flagItems = append(flagItems, ldstoretypes.KeyedItemDescriptor{
+			Key:  key,
+			Item: item,
+		})
 	}
 
+	segmentItems := make([]ldstoretypes.KeyedItemDescriptor, 0, len(t.currentSegments))
 	for key, item := range t.currentSegments {
-		json, err := json.Marshal(item.Item)
-		if err != nil {
-			return nil, err
-		}
-		builder.AddPut(subsystems.SegmentKind, key, item.Version, json)
+		segmentItems = append(segmentItems, ldstoretypes.KeyedItemDescriptor{
+			Key:  key,
+			Item: item,
+		})
 	}
 
-	return builder.Finish(subsystems.NewSelector(strconv.Itoa(version), version))
+	collections := make([]ldstoretypes.Collection, 0, 2)
+	if len(flagItems) > 0 {
+		collections = append(collections, ldstoretypes.Collection{
+			Kind:  ldstoreimpl.Features(),
+			Items: flagItems,
+		})
+	}
+	if len(segmentItems) > 0 {
+		collections = append(collections, ldstoretypes.Collection{
+			Kind:  ldstoreimpl.Segments(),
+			Items: segmentItems,
+		})
+	}
+
+	return subsystems.NewChangeSetFromCollections(
+		intent,
+		subsystems.NewSelector(strconv.Itoa(version), version),
+		collections,
+	)
 }
 
 func (t *TestDataSynchronizer) makeTransferChangesForObject(