Add retry for resource_purge_deleted call

Retry when resource_purge_deleted failed with deadlock or other exceptions.

Change-Id: Iadab7b45e653fc6f465e257d3f450c4eaf509297
Story: #2001997
Task: #19629

heat/db/sqlalchemy/api.py

@@ -234,6 +234,8 @@ def resource_get_all(context):
     return results
 
 
+@oslo_db_api.wrap_db_retry(max_retries=3, retry_on_deadlock=True,
+                           retry_interval=0.5, inc_retry_interval=True)
 def resource_purge_deleted(context, stack_id):
     filters = {'stack_id': stack_id, 'action': 'DELETE', 'status': 'COMPLETE'}
     query = context.session.query(models.Resource)

heat/tests/db/test_sqlalchemy_api.py

@@ -2516,6 +2516,19 @@ class DBAPIResourceTest(common.HeatTestCase):
         self.assertRaises(exception.NotFound, db_api.resource_get,
                           self.ctx, resource.id)
 
+    @mock.patch.object(time, 'sleep')
+    def test_resource_purge_deleted_by_stack_retry_on_deadlock(self, m_sleep):
+        val = {'name': 'res1', 'action': rsrc.Resource.DELETE,
+               'status': rsrc.Resource.COMPLETE}
+        create_resource(self.ctx, self.stack, **val)
+
+        with mock.patch('sqlalchemy.orm.query.Query.delete',
+                        side_effect=db_exception.DBDeadlock) as mock_delete:
+            self.assertRaises(db_exception.DBDeadlock,
+                              db_api.resource_purge_deleted,
+                              self.ctx, self.stack.id)
+            self.assertEqual(4, mock_delete.call_count)
+
     def test_engine_get_all_locked_by_stack(self):
         values = [
             {'name': 'res1', 'action': rsrc.Resource.DELETE,