Add InstanceVersion for our persisted types

backend/pkg/informers/informers_test.go

@@ -260,15 +260,14 @@ func subscriptionInformerTestCase() informerTestCase {
 		expectedInitialAdds: 2,
 		mutateDB: func(t *testing.T, ctx context.Context, mockResourcesDBClient *databasetesting.MockResourcesDBClient) {
 			t.Helper()
-			// Update sub-1.
-			sub1Updated := &arm.Subscription{
-				CosmosMetadata: arm.CosmosMetadata{
-					ResourceID: mustParseResourceID(t, "/subscriptions/sub-1"),
-				},
-				ResourceID: mustParseResourceID(t, "/subscriptions/sub-1"),
-				State:      arm.SubscriptionStateWarned,
-			}
-			_, err := mockResourcesDBClient.Subscriptions().Replace(ctx, sub1Updated, nil)
+			// Update sub-1. Deep-copy the existing document so the Replace
+			// carries the existing etag and instance version forward; the
+			// crud_helpers PrepareForReplace check rejects fresh-built docs.
+			existing, err := mockResourcesDBClient.Subscriptions().Get(ctx, "sub-1")
+			require.NoError(t, err)
+			sub1Updated := existing.DeepCopy()
+			sub1Updated.State = arm.SubscriptionStateWarned
+			_, err = mockResourcesDBClient.Subscriptions().Replace(ctx, sub1Updated, nil)
 			require.NoError(t, err)
 
 			// Add sub-3.
@@ -380,8 +379,14 @@ func clusterInformerTestCase() informerTestCase {
 			t.Helper()
 			clusterCRUD := mockResourcesDBClient.HCPClusters(subscriptionID, resourceGroupName)
 
-			// Update cluster-1 state.
-			_, err := clusterCRUD.Replace(ctx, newCluster(t, "cluster-1", arm.ProvisioningStateDeleting), nil)
+			// Update cluster-1 state. Deep-copy the existing document so the
+			// Replace carries the existing etag and instance version forward;
+			// the crud_helpers PrepareForReplace check rejects fresh-built docs.
+			existing, err := clusterCRUD.Get(ctx, "cluster-1")
+			require.NoError(t, err)
+			updated := existing.DeepCopy()
+			updated.ServiceProviderProperties.ProvisioningState = arm.ProvisioningStateDeleting
+			_, err = clusterCRUD.Replace(ctx, updated, nil)
 			require.NoError(t, err)
 
 			// Add cluster-3.
@@ -515,8 +520,14 @@ func nodePoolInformerTestCase() informerTestCase {
 			t.Helper()
 			npCRUD := mockResourcesDBClient.HCPClusters(subscriptionID, resourceGroupName).NodePools(clusterName)
 
-			// Update np-1 replica count.
-			_, err := npCRUD.Replace(ctx, newNodePool(t, "np-1", 10), nil)
+			// Update np-1 replica count. Deep-copy the existing document so
+			// the Replace carries the existing etag and instance version forward;
+			// the crud_helpers PrepareForReplace check rejects fresh-built docs.
+			existing, err := npCRUD.Get(ctx, "np-1")
+			require.NoError(t, err)
+			updated := existing.DeepCopy()
+			updated.Properties.Replicas = 10
+			_, err = npCRUD.Replace(ctx, updated, nil)
 			require.NoError(t, err)
 
 			// Add np-3.
@@ -618,8 +629,14 @@ func activeOperationInformerTestCase() informerTestCase {
 
 			// Transition op-1 to succeeded (terminal) — the active operations
 			// informer should see this as a deletion since it only tracks
-			// non-terminal operations.
-			_, err := opCRUD.Replace(ctx, newOperation(t, "op-1", arm.ProvisioningStateSucceeded), nil)
+			// non-terminal operations. Deep-copy the existing document so the
+			// Replace carries the existing etag and instance version forward;
+			// the crud_helpers PrepareForReplace check rejects fresh-built docs.
+			existing, err := opCRUD.Get(ctx, "op-1")
+			require.NoError(t, err)
+			updated := existing.DeepCopy()
+			updated.Status = arm.ProvisioningStateSucceeded
+			_, err = opCRUD.Replace(ctx, updated, nil)
 			require.NoError(t, err)
 
 			// Add a new active operation op-3.

frontend/pkg/frontend/frontend.go

@@ -568,6 +568,9 @@ func (f *Frontend) ArmSubscriptionPut(writer http.ResponseWriter, request *http.
 			logger.Info(message)
 		}
 		if len(messages) > 0 {
+			// Carry the etag of the just-read document forward so Replace
+			// is conditional on it; the DB layer refuses unconditional updates.
+			requestSubscription.CosmosMetadata = *existingSubscription.CosmosMetadata.DeepCopy()
 			resultingSubscription, err = f.resourcesDBClient.Subscriptions().Replace(ctx, &requestSubscription, nil)
 			if err != nil {
 				return utils.TrackError(err)

internal/api/arm/types_cosmosdata.go

@@ -35,6 +35,13 @@ type CosmosMetadata struct {
 	ExistingCosmosUID string `json:"-"`
 
 	CosmosETag azcore.ETag `json:"etag,omitempty"`
+
+	// InstanceVersion is a field that auto-increments every time the resource is updated.  This gives us the ability to
+	// compare two instances of stored resources and determine which one is newer.  We will use this field to integrate
+	// changefeeds for controllers and decide which changes are newer than the level we have already observed so that we
+	// can periodically re-list all items.
+	// The auto-incrementing happens automatically in the storage layer for conditional updates.
+	InstanceVersion int64 `json:"instanceVersion"`
 }
 
 var (

internal/api/v20240610preview/conversion_fuzz_test.go

@@ -46,6 +46,9 @@ func TestRoundTripInternalExternalInternal(t *testing.T) {
 			j.ResourceID = api.Must(azcorearm.ParseResourceID("/subscriptions/00000000-0000-0000-0000-000000000000/resourcegroups/myrg"))
 			j.ExistingCosmosUID = ""
 			j.CosmosETag = ""
+			// InstanceVersion is purely storage-layer state; it does not round-trip
+			// through the external API.
+			j.InstanceVersion = 0
 		},
 		func(j *api.HCPOpenShiftClusterCustomerProperties, c randfill.Continue) {
 			c.FillNoCustom(j)
@@ -129,22 +132,31 @@ func TestRoundTripInternalExternalInternal(t *testing.T) {
 	for i := 0; i < 200; i++ {
 		original := &api.HCPOpenShiftCluster{}
 		fuzzer.Fill(original)
+		// InstanceVersion does not roundtrip through the external type because
+		// it is purely a database concern. The CosmosMetadata fuzz override
+		// also zeroes this, but randfill does not always dispatch the custom
+		// func when filling an embedded struct in-place, so zero it here too.
+		original.InstanceVersion = 0
 		roundTripHCPCluster(t, original)
 	}
 
 	for i := 0; i < 200; i++ {
 		original := &api.HCPOpenShiftClusterNodePool{}
 		fuzzer.Fill(original)
-		// CosmosETag does not roundtrip through the external type because it is purely a database concern
+		// CosmosETag and InstanceVersion do not roundtrip through the external
+		// type because they are purely database concerns.
 		original.CosmosETag = ""
+		original.InstanceVersion = 0
 		roundTripNodePool(t, original)
 	}
 
 	for i := 0; i < 200; i++ {
 		original := &api.HCPOpenShiftClusterExternalAuth{}
 		fuzzer.Fill(original)
-		// CosmosETag does not roundtrip through the external type because it is purely a database concern
+		// CosmosETag and InstanceVersion do not roundtrip through the external
+		// type because they are purely database concerns.
 		original.CosmosETag = ""
+		original.InstanceVersion = 0
 		roundTripExternalAuth(t, original)
 	}
 }

internal/api/v20251223preview/conversion_fuzz_test.go

@@ -46,6 +46,9 @@ func TestRoundTripInternalExternalInternal(t *testing.T) {
 			j.ResourceID = api.Must(azcorearm.ParseResourceID("/subscriptions/00000000-0000-0000-0000-000000000000/resourcegroups/myrg"))
 			j.ExistingCosmosUID = ""
 			j.CosmosETag = ""
+			// InstanceVersion is purely storage-layer state; it does not round-trip
+			// through the external API.
+			j.InstanceVersion = 0
 		},
 		func(j *api.ImageDigestMirror, c randfill.Continue) {
 			c.FillNoCustom(j)
@@ -107,22 +110,31 @@ func TestRoundTripInternalExternalInternal(t *testing.T) {
 	for i := 0; i < 200; i++ {
 		original := &api.HCPOpenShiftCluster{}
 		fuzzer.Fill(original)
+		// InstanceVersion does not roundtrip through the external type because
+		// it is purely a database concern. The CosmosMetadata fuzz override
+		// also zeroes this, but randfill does not always dispatch the custom
+		// func when filling an embedded struct in-place, so zero it here too.
+		original.InstanceVersion = 0
 		roundTripHCPCluster(t, original)
 	}
 
 	for i := 0; i < 200; i++ {
 		original := &api.HCPOpenShiftClusterNodePool{}
 		fuzzer.Fill(original)
-		// CosmosETag does not roundtrip through the external type because it is purely a database concern
+		// CosmosETag and InstanceVersion do not roundtrip through the external
+		// type because they are purely database concerns.
 		original.CosmosETag = ""
+		original.InstanceVersion = 0
 		roundTripNodePool(t, original)
 	}
 
 	for i := 0; i < 200; i++ {
 		original := &api.HCPOpenShiftClusterExternalAuth{}
 		fuzzer.Fill(original)
-		// CosmosETag does not roundtrip through the external type because it is purely a database concern
+		// CosmosETag and InstanceVersion do not roundtrip through the external
+		// type because they are purely database concerns.
 		original.CosmosETag = ""
+		original.InstanceVersion = 0
 		roundTripExternalAuth(t, original)
 	}
 }

internal/database/convert_any.go

@@ -20,7 +20,6 @@ import (
 
 	azcorearm "github.com/Azure/azure-sdk-for-go/sdk/azcore/arm"
 
-	"github.com/Azure/ARO-HCP/internal/api"
 	"github.com/Azure/ARO-HCP/internal/api/fleet"
 	"github.com/Azure/ARO-HCP/internal/api/kubeapplier"
 	"github.com/Azure/ARO-HCP/internal/utils"
@@ -30,15 +29,6 @@ func CosmosToInternal[InternalAPIType, CosmosAPIType any](obj *CosmosAPIType) (*
 	var internalObj any
 	var err error
 	switch cosmosObj := any(obj).(type) {
-	case *ExternalAuth:
-		internalObj, err = CosmosToInternalExternalAuth(cosmosObj)
-
-	case *HCPCluster:
-		internalObj, err = CosmosToInternalCluster(cosmosObj)
-
-	case *NodePool:
-		internalObj, err = CosmosToInternalNodePool(cosmosObj)
-
 	case *TypedDocument:
 		var expectedObj InternalAPIType
 		switch castObj := any(expectedObj).(type) {
@@ -86,14 +76,6 @@ func InternalToCosmos[InternalAPIType, CosmosAPIType any](obj *InternalAPIType)
 	var cosmosObj any
 	var err error
 	switch internalObj := any(obj).(type) {
-	case *api.HCPOpenShiftClusterExternalAuth:
-		cosmosObj, err = InternalToCosmosExternalAuth(internalObj)
-
-	case *api.HCPOpenShiftCluster:
-		cosmosObj, err = InternalToCosmosCluster(internalObj)
-
-	case *api.HCPOpenShiftClusterNodePool:
-		cosmosObj, err = InternalToCosmosNodePool(internalObj)
 
 	case *fleet.Stamp:
 		cosmosObj, err = InternalToCosmosFleet(internalObj)