[snapshot] Elide writes for RRO with no changes (#15976)

These changes elide checkpoint writes for RegisterResourceOutputs steps
if the associated resource's outputs have not changed. This can
eliminate unnecessary snapshot writes for stacks with large numbers of
component resources whose outputs do not change.

changelog/pending/20240418--backend-diy-service--elide-state-file-writes-for-unchanged-component-outputs.yaml

@@ -0,0 +1,4 @@
+changes:
+- type: feat
+  scope: backend/diy,service
+  description: Elide state file writes for unchanged component outputs

pkg/backend/snapshot.go

@@ -124,7 +124,15 @@ func (sm *SnapshotManager) mutate(mutator func() bool) error {
 // Note that this is completely not thread-safe and defeats the purpose of having a `mutate` callback
 // entirely, but the hope is that this state of things will not be permament.
 func (sm *SnapshotManager) RegisterResourceOutputs(step deploy.Step) error {
-	return sm.mutate(func() bool { return true })
+	return sm.mutate(func() bool {
+		old, new := step.Old(), step.New()
+		if old != nil && new != nil && old.Outputs.DeepEquals(new.Outputs) {
+			logging.V(9).Infof("SnapshotManager: eliding RegisterResourceOutputs due to equal outputs")
+			return false
+		}
+
+		return true
+	})
 }
 
 // BeginMutation signals to the SnapshotManager that the engine intends to mutate the global snapshot

pkg/backend/snapshot_test.go

@@ -1043,17 +1043,25 @@ func TestRegisterOutputs(t *testing.T) {
 	manager, sp := MockSetup(t, snap)
 
 	// There should be zero snaps performed at the start.
-	assert.Len(t, sp.SavedSnapshots, 0)
+	require.Empty(t, sp.SavedSnapshots)
 
 	// The step here is not important.
 	step := deploy.NewSameStep(nil, nil, resourceA, resourceA)
 	err := manager.RegisterResourceOutputs(step)
-	if !assert.NoError(t, err) {
-		t.FailNow()
-	}
+	require.NoError(t, err)
 
-	// The RegisterResourceOutputs should have caused a snapshot to be written.
-	assert.Len(t, sp.SavedSnapshots, 1)
+	// The RegisterResourceOutputs should not have caused a snapshot to be written.
+	require.Empty(t, sp.SavedSnapshots)
+
+	// Now, change the outputs and issue another RRO.
+	resourceA2 := NewResource("a")
+	resourceA2.Outputs = resource.PropertyMap{"hello": resource.NewStringProperty("world")}
+	step = deploy.NewSameStep(nil, nil, resourceA, resourceA2)
+	err = manager.RegisterResourceOutputs(step)
+	require.NoError(t, err)
+
+	// The new outputs should have been saved.
+	require.Len(t, sp.SavedSnapshots, 1)
 
 	// It should be identical to what has already been written.
 	lastSnap := sp.LastSnap()