Merge pull request saltstack#50778 from bloomberg/saltnado-fix

saltnado: fix timeout handling

salt/netapi/rest_tornado/saltnado.py

@@ -979,6 +979,7 @@ def _disbatch_local(self, chunk):
             min_wait_time = tornado.gen.sleep(self.application.opts['syndic_wait'])
 
         # To ensure job_not_running and all_return are terminated by each other, communicate using a future
+        is_timed_out = tornado.gen.sleep(self.application.opts['gather_job_timeout'])
         is_finished = Future()
 
         # ping until the job is not running, while doing so, if we see new minions returning
@@ -1001,18 +1002,18 @@ def more_todo():
         # early if gather_job_timeout has been exceeded
         chunk_ret = {}
         while True:
-            to_wait = events+[is_finished]
+            to_wait = events+[is_finished, is_timed_out]
             if not min_wait_time.done():
                 to_wait += [min_wait_time]
 
             def cancel_inflight_futures():
                 for event in to_wait:
-                    if not event.done():
+                    if not event.done() and event is not is_timed_out:
                         event.set_result(None)
             f = yield Any(to_wait)
             try:
                 # When finished entire routine, cleanup other futures and return result
-                if f is is_finished:
+                if f is is_finished or f is is_timed_out:
                     cancel_inflight_futures()
                     raise tornado.gen.Return(chunk_ret)
                 elif f is min_wait_time:
@@ -1067,7 +1068,7 @@ def job_not_running(self, jid, tgt, tgt_type, minions, is_finished):
                 if not event.done():
                     event.set_result(None)
 
-                if not minion_running:
+                if not minion_running or is_finished.done():
                     raise tornado.gen.Return(True)
                 else:
                     ping_pub_data = yield self.saltclients['local'](tgt,