Build item policies without repeated accumulated copies

internal/engine/imports.go

@@ -960,7 +960,12 @@ func projectNestedCollectionPolicies(composed *composedDocument, scope acquisiti
 	if composed == nil {
 		return
 	}
-	baseOffsets := make(map[string]int)
+	type itemPolicyProjection struct {
+		destination Path
+		policy      collectionPolicy
+		base        int
+	}
+	projected := make(map[string]*itemPolicyProjection)
 	var prefixLengths map[string]int
 	for sourceIndex, contribution := range scope.contributions {
 		if prefixLengths != nil && sourceIndex > 0 {
@@ -983,17 +988,21 @@ func projectNestedCollectionPolicies(composed *composedDocument, scope acquisiti
 			}
 			itemIndex = prefixLengths[key]
 		}
-		base, initialized := baseOffsets[key]
-		policy, _ := composed.collectionPolicies.at(contribution.destination)
+		working, initialized := projected[key]
 		if !initialized {
-			base = len(policy.recordOrder)
-			baseOffsets[key] = base
+			// Read one owned copy per destination, then add each incoming
+			// record without recopying earlier records and their scopes.
+			policy, _ := composed.collectionPolicies.at(contribution.destination)
+			working = &itemPolicyProjection{destination: contribution.destination, policy: policy, base: len(policy.recordOrder)}
+			projected[key] = working
 		}
-		policy.setRecordPolicyAt(base+itemIndex, collectionRecordPolicy{
+		working.policy.setRecordPolicyAt(working.base+itemIndex, collectionRecordPolicy{
 			owner: newCollectionRecordPolicy().owner,
 			scope: contribution.nested.collectionPolicies,
 		})
-		composed.collectionPolicies.set(contribution.destination, policy)
+	}
+	for _, working := range projected {
+		composed.collectionPolicies.set(working.destination, working.policy)
 	}
 }
 

internal/engine/item_policy_test.go

@@ -229,6 +229,64 @@ func TestItemPolicyProjectionCountsContributionsWithoutLocalPolicy(t *testing.T)
 	}
 }
 
+func TestItemPolicyProjectionDetachesRepeatedScopes(t *testing.T) {
+	t.Parallel()
+	local := collectionPolicyScope{}
+	local.set(Keys("steps"), collectionPolicy{identityFields: []string{"id"}, identityPresent: true})
+	scope := acquisitionScope{}
+	ranges := make(map[sequenceRangeKey]sequenceRange)
+	for index, destination := range []string{"items", "other", "items"} {
+		path := Keys(destination)
+		scope.contributions = append(scope.contributions, acquiredContribution{
+			document: parseItemPolicyDocument(t, "item", destination+": [{}]\n"), destination: path,
+			nested: &composedDocument{collectionPolicies: local},
+		})
+		output := 0
+		if index == 2 {
+			output = 1
+		}
+		ranges[sequenceRangeKey{sourceIndex: index, path: path.String()}] = sequenceRange{start: output, end: output + 1}
+	}
+	first, second := &composedDocument{}, &composedDocument{}
+	projectNestedCollectionPolicies(first, scope, ranges)
+	projectNestedCollectionPolicies(second, scope, ranges)
+	policy := first.collectionPolicies.collections[Keys("items").String()].policy
+	if len(policy.recordOrder) != 2 || policy.recordOrder[0] == policy.recordOrder[1] {
+		t.Fatal("separately placed records did not retain distinct owners")
+	}
+	other := second.collectionPolicies.collections[Keys("items").String()].policy
+	if policy.recordOrder[0] == other.recordOrder[0] {
+		t.Fatal("independent projections reused a record owner")
+	}
+	// Mutate stored state directly: a detached query result would conceal an
+	// accidental alias between incrementally constructed record scopes.
+	steps := Keys("steps").String()
+	firstRecord := policy.recordPolicies[policy.recordOrder[0]]
+	firstRecord.collections[steps].policy.identityFields[0] = "changed"
+	for _, check := range []struct {
+		composed *composedDocument
+		path     Path
+		index    int
+	}{
+		{first, Keys("items"), 1}, {first, Keys("other"), 0},
+		{second, Keys("items"), 0}, {second, Keys("items"), 1}, {second, Keys("other"), 0},
+	} {
+		outer, _ := check.composed.collectionPolicies.at(check.path)
+		record, found := outer.recordPolicyAt(check.index)
+		identity, present := record.scope.at(Keys("steps"))
+		if !found || !present || !reflect.DeepEqual(identity.identityFields, []string{"id"}) {
+			t.Fatalf("mutation leaked into %s item %d: %#v", check.path, check.index, identity)
+		}
+	}
+	if local.collections[steps].policy.identityFields[0] != "id" {
+		t.Fatal("projected state mutated the incoming scope")
+	}
+	local.collections[steps].policy.identityFields[0] = "source-changed"
+	if other.recordPolicies[other.recordOrder[0]].collections[steps].policy.identityFields[0] != "id" {
+		t.Fatal("incoming scope remained aliased to projected state")
+	}
+}
+
 func TestOverlaySequenceTransfersIncomingItemPolicyOwnership(t *testing.T) {
 	path := Keys("records")
 	registered := mustCompileSequenceIdentityPolicy(t, []sequenceIdentityRegistration{{selector: sequenceIdentitySelector{literalSequenceSelector("records")}, identityFields: []string{"id"}}})