Don't unsubscribe from open events on the CLI too early on long-running commands

Fixes #32305

salt/client/__init__.py

@@ -21,17 +21,10 @@ The data structure needs to be:
 # Import python libs
 from __future__ import absolute_import, print_function
 import os
-import sys
 import time
-import copy
-import errno
 import logging
-import re
 from datetime import datetime
 
-# Import 3rd-party libs
-
-
 # Import salt libs
 import salt.config
 import salt.minion
@@ -219,7 +212,7 @@ class LocalClient(object):
         Common checks on the pub_data data structure returned from running pub
         '''
         if not pub_data:
-            # Failed to autnenticate, this could be a bunch of things
+            # Failed to authenticate, this could be a bunch of things
             raise EauthAuthenticationError(
                 'Failed to authenticate! This is most likely because this '
                 'user is not permitted to execute commands, but there is a '
@@ -293,7 +286,7 @@ class LocalClient(object):
                 'The salt master could not be contacted. Is master running?'
             )
         except Exception as general_exception:
-            # Convert to generic client error and pass along mesasge
+            # Convert to generic client error and pass along message
             raise SaltClientError(general_exception)
 
         return self._check_pub_data(pub_data)
@@ -573,7 +566,7 @@ class LocalClient(object):
             **kwargs):
         '''
         Used by the :command:`salt` CLI. This method returns minion returns as
-        the come back and attempts to block until all minions return.
+        they come back and attempts to block until all minions return.
 
         The function signature is the same as :py:meth:`cmd` with the
         following exceptions.
@@ -876,6 +869,8 @@ class LocalClient(object):
             ret_iter = self.get_returns_no_block('salt/job/{0}'.format(jid))
         # iterator for the info of this job
         jinfo_iter = []
+        # open event jids that need to be un-subscribed from later
+        open_jids = set()
         timeout_at = time.time() + timeout
         gather_syndic_wait = time.time() + self.opts['syndic_wait']
         # are there still minions running the job out there
@@ -956,10 +951,11 @@ class LocalClient(object):
             for raw in jinfo_iter:
                 # if there are no more events, lets stop waiting for the jinfo
                 if raw is None:
-                    self.event.unsubscribe(jinfo['jid'])
-                    jinfo_iter = []
                     break
 
+                # Keep track of the jid events to unsubscribe from later
+                open_jids.add(jinfo['jid'])
+
                 # TODO: move to a library??
                 if 'minions' in raw.get('data', {}):
                     minions.update(raw['data']['minions'])
@@ -1002,6 +998,12 @@ class LocalClient(object):
                 time.sleep(0.01)
             else:
                 yield
+
+        # If there are any remaining open events, clean them up.
+        if open_jids:
+            for jid in open_jids:
+                self.event.unsubscribe(jid)
+
         if expect_minions:
             for minion in list((minions - found)):
                 yield {minion: {'failed': True}}