diff --git a/heat/engine/service.py b/heat/engine/service.py
index b528f243a9..0ba2ef183a 100644
--- a/heat/engine/service.py
+++ b/heat/engine/service.py
@@ -1377,7 +1377,6 @@ class EngineService(service.ServiceBase):
             def convergence_delete():
                 stack.thread_group_mgr = self.thread_group_mgr
                 self.worker_service.stop_all_workers(stack)
-                stack.delete_all_snapshots()
                 template = templatem.Template.create_empty_template(
                     from_template=stack.t)
                 stack.converge_stack(template=template, action=stack.DELETE)
diff --git a/heat/engine/stack.py b/heat/engine/stack.py
index a08d1e04ce..6a1e954b9a 100644
--- a/heat/engine/stack.py
+++ b/heat/engine/stack.py
@@ -302,16 +302,18 @@ class Stack(collections.Mapping):
         return {n: self.defn.output_definition(n)
                 for n in self.defn.enabled_output_names()}
 
+    def _resources_for_defn(self, stack_defn):
+        return {
+            name: resource.Resource(name,
+                                    stack_defn.resource_definition(name),
+                                    self)
+            for name in stack_defn.enabled_rsrc_names()
+        }
+
     @property
     def resources(self):
         if self._resources is None:
-            self._resources = {
-                name: resource.Resource(name,
-                                        self.defn.resource_definition(name),
-                                        self)
-                for name in self.defn.enabled_rsrc_names()
-            }
-
+            self._resources = self._resources_for_defn(self.defn)
         return self._resources
 
     def _update_all_resource_data(self, for_resources, for_outputs):
@@ -1371,6 +1373,14 @@ class Stack(collections.Mapping):
                          'action': self.action})
             return
 
+        if self.action == self.DELETE:
+            try:
+                self.delete_all_snapshots()
+            except Exception as exc:
+                self.state_set(self.action, self.FAILED, six.text_type(exc))
+                self.purge_db()
+                return
+
         LOG.debug('Starting traversal %s with dependencies: %s',
                   self.current_traversal, self.convergence_dependencies)
 
@@ -1980,21 +1990,28 @@ class Stack(collections.Mapping):
             self.delete_snapshot(snapshot)
             snapshot_object.Snapshot.delete(self.context, snapshot.id)
 
+    @staticmethod
+    def _template_from_snapshot_data(snapshot_data):
+        env = environment.Environment(snapshot_data['environment'])
+        files = snapshot_data['files']
+        return tmpl.Template(snapshot_data['template'], env=env, files=files)
+
     @profiler.trace('Stack.delete_snapshot', hide_args=False)
     def delete_snapshot(self, snapshot):
         """Remove a snapshot from the backends."""
-        for name, rsrc in six.iteritems(self.resources):
-            snapshot_data = snapshot.data
-            if snapshot_data:
+        snapshot_data = snapshot.data
+        if snapshot_data:
+            template = self._template_from_snapshot_data(snapshot_data)
+            ss_defn = self.defn.clone_with_new_template(template,
+                                                        self.identifier())
+            resources = self._resources_for_defn(ss_defn)
+            for name, rsrc in six.iteritems(resources):
                 data = snapshot.data['resources'].get(name)
                 if data:
                     scheduler.TaskRunner(rsrc.delete_snapshot, data)()
 
     def restore_data(self, snapshot):
-        env = environment.Environment(snapshot.data['environment'])
-        files = snapshot.data['files']
-        template = tmpl.Template(snapshot.data['template'],
-                                 env=env, files=files)
+        template = self._template_from_snapshot_data(snapshot.data)
         newstack = self.__class__(self.context, self.name, template,
                                   timeout_mins=self.timeout_mins,
                                   disable_rollback=self.disable_rollback)
diff --git a/heat/tests/test_convg_stack.py b/heat/tests/test_convg_stack.py
index df24fb2e9d..8e6c5d87fe 100644
--- a/heat/tests/test_convg_stack.py
+++ b/heat/tests/test_convg_stack.py
@@ -21,6 +21,7 @@ from heat.engine import stack as parser
 from heat.engine import template as templatem
 from heat.objects import raw_template as raw_template_object
 from heat.objects import resource as resource_objects
+from heat.objects import snapshot as snapshot_objects
 from heat.objects import stack as stack_object
 from heat.objects import sync_point as sync_point_object
 from heat.rpc import worker_client
@@ -509,6 +510,37 @@ class StackConvergenceCreateUpdateDeleteTest(common.HeatTestCase):
         self.assertTrue(mock_syncpoint_del.called)
         self.assertTrue(mock_ccu.called)
 
+    def test_snapshot_delete(self, mock_cr):
+        tmpl = {'HeatTemplateFormatVersion': '2012-12-12',
+                'Resources': {'R1': {'Type': 'GenericResourceType'}}}
+        stack = parser.Stack(utils.dummy_context(), 'updated_time_test',
+                             templatem.Template(tmpl))
+        stack.current_traversal = 'prev_traversal'
+        stack.action, stack.status = stack.CREATE, stack.COMPLETE
+        stack.store()
+        stack.thread_group_mgr = tools.DummyThreadGroupManager()
+        snapshot_values = {
+            'stack_id': stack.id,
+            'name': 'fake_snapshot',
+            'tenant': stack.context.tenant_id,
+            'status': 'COMPLETE',
+            'data': None
+        }
+        snapshot_objects.Snapshot.create(stack.context, snapshot_values)
+
+        # Ensure that snapshot is not deleted on stack update
+        stack.converge_stack(template=stack.t, action=stack.UPDATE)
+        db_snapshot_obj = snapshot_objects.Snapshot.get_all(
+            stack.context, stack.id)
+        self.assertEqual('fake_snapshot', db_snapshot_obj[0].name)
+        self.assertEqual(stack.id, db_snapshot_obj[0].stack_id)
+
+        # Ensure that snapshot is deleted on stack delete
+        stack.converge_stack(template=stack.t, action=stack.DELETE)
+        self.assertEqual([], snapshot_objects.Snapshot.get_all(
+            stack.context, stack.id))
+        self.assertTrue(mock_cr.called)
+
 
 @mock.patch.object(parser.Stack, '_persist_state')
 class TestConvgStackStateSet(common.HeatTestCase):