diff --git a/heat/engine/service.py b/heat/engine/service.py
index 0ba2ef183a..0101e5e3cf 100644
--- a/heat/engine/service.py
+++ b/heat/engine/service.py
@@ -1374,14 +1374,19 @@ class EngineService(service.ServiceBase):
         self.resource_enforcer.enforce_stack(stack, is_registered_policy=True)
 
         if stack.convergence and cfg.CONF.convergence_engine:
-            def convergence_delete():
-                stack.thread_group_mgr = self.thread_group_mgr
-                self.worker_service.stop_all_workers(stack)
-                template = templatem.Template.create_empty_template(
-                    from_template=stack.t)
-                stack.converge_stack(template=template, action=stack.DELETE)
+            stack.thread_group_mgr = self.thread_group_mgr
+            template = templatem.Template.create_empty_template(
+                from_template=stack.t)
 
-            self.thread_group_mgr.start(stack.id, convergence_delete)
+            # stop existing traversal; mark stack as FAILED
+            if stack.status == stack.IN_PROGRESS:
+                self.worker_service.stop_traversal(stack)
+
+            def stop_workers():
+                self.worker_service.stop_all_workers(stack)
+
+            stack.converge_stack(template=template, action=stack.DELETE,
+                                 pre_converge=stop_workers)
             return
 
         lock = stack_lock.StackLock(cnxt, stack.id, self.engine_id)
diff --git a/heat/engine/stack.py b/heat/engine/stack.py
index 6a1e954b9a..7dab91a7d1 100644
--- a/heat/engine/stack.py
+++ b/heat/engine/stack.py
@@ -1300,7 +1300,8 @@ class Stack(collections.Mapping):
         updater()
 
     @profiler.trace('Stack.converge_stack', hide_args=False)
-    def converge_stack(self, template, action=UPDATE, new_stack=None):
+    def converge_stack(self, template, action=UPDATE, new_stack=None,
+                       pre_converge=None):
         """Update the stack template and trigger convergence for resources."""
         if action not in [self.CREATE, self.ADOPT]:
             # no back-up template for create action
@@ -1354,9 +1355,10 @@ class Stack(collections.Mapping):
 
         # TODO(later): lifecycle_plugin_utils.do_pre_ops
 
-        self.thread_group_mgr.start(self.id, self._converge_create_or_update)
+        self.thread_group_mgr.start(self.id, self._converge_create_or_update,
+                                    pre_converge=pre_converge)
 
-    def _converge_create_or_update(self):
+    def _converge_create_or_update(self, pre_converge=None):
         current_resources = self._update_or_store_resources()
         self._compute_convg_dependencies(self.ext_rsrcs_db, self.dependencies,
                                          current_resources)
@@ -1373,6 +1375,8 @@ class Stack(collections.Mapping):
                          'action': self.action})
             return
 
+        if callable(pre_converge):
+            pre_converge()
         if self.action == self.DELETE:
             try:
                 self.delete_all_snapshots()
diff --git a/heat/engine/worker.py b/heat/engine/worker.py
index 2e38e7acb5..274dd436fe 100644
--- a/heat/engine/worker.py
+++ b/heat/engine/worker.py
@@ -125,11 +125,11 @@ class WorkerService(object):
                 _stop_traversal(child)
 
     def stop_all_workers(self, stack):
-        # stop the traversal
-        if stack.status == stack.IN_PROGRESS:
-            self.stop_traversal(stack)
+        """Cancel all existing worker threads for the stack.
 
-        # cancel existing workers
+        Threads will stop running at their next yield point, whether or not the
+        resource operations are complete.
+        """
         cancelled = _cancel_workers(stack, self.thread_group_mgr,
                                     self.engine_id, self._rpc_client)
         if not cancelled:
diff --git a/heat/tests/convergence/framework/worker_wrapper.py b/heat/tests/convergence/framework/worker_wrapper.py
index 042bc9e07d..7ca39ada1d 100644
--- a/heat/tests/convergence/framework/worker_wrapper.py
+++ b/heat/tests/convergence/framework/worker_wrapper.py
@@ -37,5 +37,8 @@ class Worker(message_processor.MessageProcessor):
                                  adopt_stack_data,
                                  converge)
 
+    def stop_traversal(self, current_stack):
+        pass
+
     def stop_all_workers(self, current_stack):
         pass
diff --git a/heat/tests/engine/test_engine_worker.py b/heat/tests/engine/test_engine_worker.py
index eeae9e3261..affb511860 100644
--- a/heat/tests/engine/test_engine_worker.py
+++ b/heat/tests/engine/test_engine_worker.py
@@ -209,7 +209,7 @@ class WorkerServiceTest(common.HeatTestCase):
         stack.id = 'stack_id'
         stack.rollback = mock.MagicMock()
         _worker.stop_all_workers(stack)
-        mock_st.assert_called_once_with(stack)
+        mock_st.assert_not_called()
         mock_cw.assert_called_once_with(stack, mock_tgm, 'engine-001',
                                         _worker._rpc_client)
         self.assertFalse(stack.rollback.called)
diff --git a/releasenotes/notes/convergence-delete-race-5b821bbd4c5ba5dc.yaml b/releasenotes/notes/convergence-delete-race-5b821bbd4c5ba5dc.yaml
new file mode 100644
index 0000000000..f06a906191
--- /dev/null
+++ b/releasenotes/notes/convergence-delete-race-5b821bbd4c5ba5dc.yaml
@@ -0,0 +1,11 @@
+---
+fixes:
+  - |
+    Previously, when deleting a convergence stack, the API call would return
+    immediately, so that it was possible for a client immediately querying the
+    status of the stack to see the state of the previous operation in progress
+    or having failed, and confuse that with a current status. (This included
+    Heat itself when acting as a client for a nested stack.) Convergence stacks
+    are now guaranteed to have moved to the ``DELETE_IN_PROGRESS`` state before
+    the delete API call returns, so any subsequent polling will reflect
+    up-to-date information.