fix(s3-deployment): physical id not set during failure scenario (#24428)

Send the correct physical id to CloudFormation to prevent unintended object deletion during rollback.

Closes #22670 

----

*By submitting this pull request, I confirm that my contribution is made under the terms of the Apache-2.0 license*

Author: garyaparker

Diff for: packages/@aws-cdk/aws-s3-deployment/lib/lambda/index.py

@@ -28,7 +28,8 @@ def handler(event, context):
 
     def cfn_error(message=None):
         logger.error("| cfn_error: %s" % message)
-        cfn_send(event, context, CFN_FAILED, reason=message)
+        cfn_send(event, context, CFN_FAILED, reason=message, physicalResourceId=event.get('PhysicalResourceId', None))
+
 
     try:
         # We are not logging ResponseURL as this is a pre-signed S3 URL, and could be used to tamper

Diff for: packages/@aws-cdk/aws-s3-deployment/test/lambda/test.py

@@ -595,6 +595,23 @@ def mock_make_api_call(self, operation_name, kwarg):
 
         self.assertEqual(update_resp['Reason'], "invalid request: request type is 'Delete' but 'PhysicalResourceId' is not defined")
 
+    def test_physical_id_on_cloud_front_error(self):
+        def mock_make_api_call(self, operation_name, kwarg):
+            if operation_name == 'CreateInvalidation':
+                raise ClientError({'Error': {'Code': '500', 'Message': 'Invalidation error'}}, operation_name)
+
+        with patch('botocore.client.BaseClient._make_api_call', new=mock_make_api_call):
+            resp = invoke_handler("Update", {
+                "SourceBucketNames": ["<source-bucket>"],
+                "SourceObjectKeys": ["<source-object-key>"],
+                "DestinationBucketName": "<dest-bucket-name>",
+                "DistributionId": "<cf-dist-id>",
+            }, old_resource_props={
+                "DestinationBucketName": "<dest-bucket-name>",
+            }, physical_id="<physical-id>", expected_status="FAILED")
+
+        self.assertEqual(resp["PhysicalResourceId"], "<physical-id>")
+
     # no bucket tags removes content
     def test_no_tags_on_bucket(self):
         def mock_make_api_call(self, operation_name, kwarg):