diff --git a/python/ray/autoscaler/autoscaler.py b/python/ray/autoscaler/autoscaler.py
index 354081fdc..1ab04d0aa 100644
--- a/python/ray/autoscaler/autoscaler.py
+++ b/python/ray/autoscaler/autoscaler.py
@@ -2,10 +2,14 @@ from __future__ import absolute_import
 from __future__ import division
 from __future__ import print_function
 
+import copy
 import json
 import hashlib
+import math
 import os
+import queue
 import subprocess
+import threading
 import time
 import traceback
 
@@ -16,8 +20,8 @@ import numpy as np
 import yaml
 
 from ray.ray_constants import AUTOSCALER_MAX_NUM_FAILURES, \
-    AUTOSCALER_MAX_CONCURRENT_LAUNCHES, AUTOSCALER_UPDATE_INTERVAL_S, \
-    AUTOSCALER_HEARTBEAT_TIMEOUT_S
+    AUTOSCALER_MAX_LAUNCH_BATCH, AUTOSCALER_MAX_CONCURRENT_LAUNCHES,\
+    AUTOSCALER_UPDATE_INTERVAL_S, AUTOSCALER_HEARTBEAT_TIMEOUT_S
 from ray.autoscaler.node_provider import get_node_provider, \
     get_default_config
 from ray.autoscaler.updater import NodeUpdaterProcess
@@ -199,6 +203,64 @@ class LoadMetrics(object):
         }
 
 
+class NodeLauncher(threading.Thread):
+    def __init__(self, queue, pending, *args, **kwargs):
+        self.queue = queue
+        self.pending = pending
+        self.provider = None
+        super(NodeLauncher, self).__init__(*args, **kwargs)
+
+    def _launch_node(self, config, count):
+        if self.provider is None:
+            self.provider = get_node_provider(config["provider"],
+                                              config["cluster_name"])
+
+        tag_filters = {TAG_RAY_NODE_TYPE: "worker"}
+        before = self.provider.nodes(tag_filters=tag_filters)
+        launch_hash = hash_launch_conf(config["worker_nodes"], config["auth"])
+        self.provider.create_node(
+            config["worker_nodes"], {
+                TAG_RAY_NODE_NAME: "ray-{}-worker".format(
+                    config["cluster_name"]),
+                TAG_RAY_NODE_TYPE: "worker",
+                TAG_RAY_NODE_STATUS: "uninitialized",
+                TAG_RAY_LAUNCH_CONFIG: launch_hash,
+            }, count)
+        after = self.provider.nodes(tag_filters=tag_filters)
+        if set(after).issubset(before):
+            print("Warning: No new nodes reported after node creation")
+
+    def run(self):
+        while True:
+            config, count = self.queue.get()
+            try:
+                self._launch_node(config, count)
+            finally:
+                self.pending.dec(count)
+
+
+class ConcurrentCounter():
+    def __init__(self):
+        self._value = 0
+        self._lock = threading.Lock()
+
+    def inc(self, count):
+        with self._lock:
+            self._value += count
+            return self._value
+
+    def dec(self, count):
+        with self._lock:
+            assert self._value >= count, "counter cannot go negative"
+            self._value -= count
+            return self._value
+
+    @property
+    def value(self):
+        with self._lock:
+            return self._value
+
+
 class StandardAutoscaler(object):
     """The autoscaling control loop for a Ray cluster.
 
@@ -220,6 +282,7 @@ class StandardAutoscaler(object):
     def __init__(self,
                  config_path,
                  load_metrics,
+                 max_launch_batch=AUTOSCALER_MAX_LAUNCH_BATCH,
                  max_concurrent_launches=AUTOSCALER_MAX_CONCURRENT_LAUNCHES,
                  max_failures=AUTOSCALER_MAX_NUM_FAILURES,
                  process_runner=subprocess,
@@ -233,6 +296,7 @@ class StandardAutoscaler(object):
                                           self.config["cluster_name"])
 
         self.max_failures = max_failures
+        self.max_launch_batch = max_launch_batch
         self.max_concurrent_launches = max_concurrent_launches
         self.verbose_updates = verbose_updates
         self.process_runner = process_runner
@@ -246,6 +310,17 @@ class StandardAutoscaler(object):
         self.last_update_time = 0.0
         self.update_interval_s = update_interval_s
 
+        # Node launchers
+        self.launch_queue = queue.Queue()
+        self.num_launches_pending = ConcurrentCounter()
+        max_batches = math.ceil(
+            max_concurrent_launches / float(max_launch_batch))
+        for i in range(int(max_batches)):
+            node_launcher = NodeLauncher(
+                queue=self.launch_queue, pending=self.num_launches_pending)
+            node_launcher.daemon = True
+            node_launcher.start()
+
         # Expand local file_mounts to allow ~ in the paths. This can't be done
         # earlier when the config is written since we might be on different
         # platform and the expansion would result in wrong path.
@@ -278,6 +353,7 @@ class StandardAutoscaler(object):
             return
 
         self.last_update_time = time.time()
+        num_pending = self.num_launches_pending.value
         nodes = self.workers()
         print(self.debug_string(nodes))
         self.load_metrics.prune_active_ips(
@@ -318,9 +394,11 @@ class StandardAutoscaler(object):
 
         # Launch new nodes if needed
         target_num = self.target_num_workers()
-        if len(nodes) < target_num:
-            self.launch_new_node(
-                min(self.max_concurrent_launches, target_num - len(nodes)))
+        num_nodes = len(nodes) + num_pending
+        if num_nodes < target_num:
+            max_allowed = min(self.max_launch_batch,
+                              self.max_concurrent_launches - num_pending)
+            self.launch_new_node(min(max_allowed, target_num - num_nodes))
             print(self.debug_string())
 
         # Process any completed updates
@@ -453,27 +531,19 @@ class StandardAutoscaler(object):
 
     def launch_new_node(self, count):
         print("StandardAutoscaler: Launching {} new nodes".format(count))
-        num_before = len(self.workers())
-        self.provider.create_node(
-            self.config["worker_nodes"], {
-                TAG_RAY_NODE_NAME: "ray-{}-worker".format(
-                    self.config["cluster_name"]),
-                TAG_RAY_NODE_TYPE: "worker",
-                TAG_RAY_NODE_STATUS: "uninitialized",
-                TAG_RAY_LAUNCH_CONFIG: self.launch_hash,
-            }, count)
-        if len(self.workers()) <= num_before:
-            print("Warning: Num nodes failed to increase after node creation")
+        self.num_launches_pending.inc(count)
+        config = copy.deepcopy(self.config)
+        self.launch_queue.put((config, count))
 
     def workers(self):
-        return self.provider.nodes(tag_filters={
-            TAG_RAY_NODE_TYPE: "worker",
-        })
+        return self.provider.nodes(tag_filters={TAG_RAY_NODE_TYPE: "worker"})
 
     def debug_string(self, nodes=None):
         if nodes is None:
             nodes = self.workers()
         suffix = ""
+        if self.num_launches_pending:
+            suffix += " ({} pending)".format(self.num_launches_pending.value)
         if self.updaters:
             suffix += " ({} updating)".format(len(self.updaters))
         if self.num_failed_updates:
diff --git a/python/ray/ray_constants.py b/python/ray/ray_constants.py
index 7e5df9650..f8d2dfe2e 100644
--- a/python/ray/ray_constants.py
+++ b/python/ray/ray_constants.py
@@ -16,6 +16,11 @@ def env_integer(key, default):
 # is a safety feature to prevent e.g. runaway node launches.
 AUTOSCALER_MAX_NUM_FAILURES = env_integer("AUTOSCALER_MAX_NUM_FAILURES", 5)
 
+# The maximum number of nodes to launch in a single request.
+# Multiple requests may be made for this batch size, up to
+# the limit of AUTOSCALER_MAX_CONCURRENT_LAUNCHES.
+AUTOSCALER_MAX_LAUNCH_BATCH = env_integer("AUTOSCALER_MAX_LAUNCH_BATCH", 5)
+
 # Max number of nodes to launch at a time.
 AUTOSCALER_MAX_CONCURRENT_LAUNCHES = env_integer(
     "AUTOSCALER_MAX_CONCURRENT_LAUNCHES", 10)
diff --git a/test/autoscaler_test.py b/test/autoscaler_test.py
index ea317aac8..9120b332e 100644
--- a/test/autoscaler_test.py
+++ b/test/autoscaler_test.py
@@ -4,6 +4,7 @@ from __future__ import print_function
 
 import shutil
 import tempfile
+import threading
 import time
 import unittest
 import yaml
@@ -50,6 +51,8 @@ class MockProvider(NodeProvider):
         self.next_id = 0
         self.throw = False
         self.fail_creates = False
+        self.ready_to_create = threading.Event()
+        self.ready_to_create.set()
 
     def nodes(self, tag_filters):
         if self.throw:
@@ -75,6 +78,7 @@ class MockProvider(NodeProvider):
         return self.mock_nodes[node_id].external_ip
 
     def create_node(self, node_config, tags, count):
+        self.ready_to_create.wait()
         if self.fail_creates:
             return
         for _ in range(count):
@@ -182,6 +186,20 @@ class AutoscalingTest(unittest.TestCase):
             time.sleep(.1)
         raise Exception("Timed out waiting for {}".format(condition))
 
+    def waitForNodes(self, expected, comparison=None, tag_filters={}):
+        MAX_ITER = 50
+        for i in range(MAX_ITER):
+            n = len(self.provider.nodes(tag_filters))
+            if comparison is None:
+                comparison = self.assertEqual
+            try:
+                comparison(n, expected)
+                return
+            except Exception:
+                if i == MAX_ITER - 1:
+                    raise
+            time.sleep(.1)
+
     def create_provider(self, config, cluster_name):
         assert self.provider
         return self.provider
@@ -241,9 +259,9 @@ class AutoscalingTest(unittest.TestCase):
             config_path, LoadMetrics(), max_failures=0, update_interval_s=0)
         self.assertEqual(len(self.provider.nodes({})), 0)
         autoscaler.update()
-        self.assertEqual(len(self.provider.nodes({})), 2)
+        self.waitForNodes(2)
         autoscaler.update()
-        self.assertEqual(len(self.provider.nodes({})), 2)
+        self.waitForNodes(2)
 
     def testTerminateOutdatedNodesGracefully(self):
         config = SMALL_CLUSTER.copy()
@@ -254,16 +272,16 @@ class AutoscalingTest(unittest.TestCase):
         self.provider.create_node({}, {TAG_RAY_NODE_TYPE: "worker"}, 10)
         autoscaler = StandardAutoscaler(
             config_path, LoadMetrics(), max_failures=0, update_interval_s=0)
-        self.assertEqual(len(self.provider.nodes({})), 10)
+        self.waitForNodes(10)
 
         # Gradually scales down to meet target size, never going too low
         for _ in range(10):
             autoscaler.update()
-            self.assertLessEqual(len(self.provider.nodes({})), 5)
-            self.assertGreaterEqual(len(self.provider.nodes({})), 4)
+            self.waitForNodes(5, comparison=self.assertLessEqual)
+            self.waitForNodes(4, comparison=self.assertGreaterEqual)
 
         # Eventually reaches steady state
-        self.assertEqual(len(self.provider.nodes({})), 5)
+        self.waitForNodes(5)
 
     def testDynamicScaling(self):
         config_path = self.write_config(SMALL_CLUSTER)
@@ -271,12 +289,52 @@ class AutoscalingTest(unittest.TestCase):
         autoscaler = StandardAutoscaler(
             config_path,
             LoadMetrics(),
+            max_launch_batch=5,
+            max_concurrent_launches=5,
+            max_failures=0,
+            update_interval_s=0)
+        self.waitForNodes(0)
+        autoscaler.update()
+        self.waitForNodes(2)
+
+        # Update the config to reduce the cluster size
+        new_config = SMALL_CLUSTER.copy()
+        new_config["max_workers"] = 1
+        self.write_config(new_config)
+        autoscaler.update()
+        self.waitForNodes(1)
+
+        # Update the config to reduce the cluster size
+        new_config["min_workers"] = 10
+        new_config["max_workers"] = 10
+        self.write_config(new_config)
+        autoscaler.update()
+        self.waitForNodes(6)
+        autoscaler.update()
+        self.waitForNodes(10)
+
+    def testDelayedLaunch(self):
+        config_path = self.write_config(SMALL_CLUSTER)
+        self.provider = MockProvider()
+        autoscaler = StandardAutoscaler(
+            config_path,
+            LoadMetrics(),
+            max_launch_batch=5,
             max_concurrent_launches=5,
             max_failures=0,
             update_interval_s=0)
         self.assertEqual(len(self.provider.nodes({})), 0)
+
+        # Update will try to create, but will block until we set the flag
+        self.provider.ready_to_create.clear()
         autoscaler.update()
-        self.assertEqual(len(self.provider.nodes({})), 2)
+        self.assertEqual(autoscaler.num_launches_pending.value, 2)
+        self.assertEqual(len(self.provider.nodes({})), 0)
+
+        # Set the flag, check it updates
+        self.provider.ready_to_create.set()
+        self.waitForNodes(2)
+        self.assertEqual(autoscaler.num_launches_pending.value, 0)
 
         # Update the config to reduce the cluster size
         new_config = SMALL_CLUSTER.copy()
@@ -285,14 +343,61 @@ class AutoscalingTest(unittest.TestCase):
         autoscaler.update()
         self.assertEqual(len(self.provider.nodes({})), 1)
 
-        # Update the config to reduce the cluster size
-        new_config["min_workers"] = 10
-        new_config["max_workers"] = 10
-        self.write_config(new_config)
+    def testDelayedLaunchWithFailure(self):
+        config = SMALL_CLUSTER.copy()
+        config["min_workers"] = 10
+        config["max_workers"] = 10
+        config_path = self.write_config(config)
+        self.provider = MockProvider()
+        autoscaler = StandardAutoscaler(
+            config_path,
+            LoadMetrics(),
+            max_launch_batch=5,
+            max_concurrent_launches=8,
+            max_failures=0,
+            update_interval_s=0)
+        self.assertEqual(len(self.provider.nodes({})), 0)
+
+        # update() should launch a wave of 5 nodes (max_launch_batch)
+        # Force this first wave to block.
+        rtc1 = self.provider.ready_to_create
+        rtc1.clear()
         autoscaler.update()
-        self.assertEqual(len(self.provider.nodes({})), 6)
+        # Synchronization: wait for launchy thread to be blocked on rtc1
+        if hasattr(rtc1, '_cond'):  # Python 3.5
+            waiters = rtc1._cond._waiters
+        else:  # Python 2.7
+            waiters = rtc1._Event__cond._Condition__waiters
+        self.waitFor(lambda: len(waiters) == 1)
+        self.assertEqual(autoscaler.num_launches_pending.value, 5)
+        self.assertEqual(len(self.provider.nodes({})), 0)
+
+        # Call update() to launch a second wave of 3 nodes,
+        # as 5 + 3 = 8 = max_concurrent_launches.
+        # Make this wave complete immediately.
+        rtc2 = threading.Event()
+        self.provider.ready_to_create = rtc2
+        rtc2.set()
         autoscaler.update()
-        self.assertEqual(len(self.provider.nodes({})), 10)
+        self.waitForNodes(3)
+        self.assertEqual(autoscaler.num_launches_pending.value, 5)
+
+        # The first wave of 5 will now tragically fail
+        self.provider.fail_creates = True
+        rtc1.set()
+        self.waitFor(lambda: autoscaler.num_launches_pending.value == 0)
+        self.assertEqual(len(self.provider.nodes({})), 3)
+
+        # Retry the first wave, allowing it to succeed this time
+        self.provider.fail_creates = False
+        autoscaler.update()
+        self.waitForNodes(8)
+        self.assertEqual(autoscaler.num_launches_pending.value, 0)
+
+        # Final wave of 2 nodes
+        autoscaler.update()
+        self.waitForNodes(10)
+        self.assertEqual(autoscaler.num_launches_pending.value, 0)
 
     def testUpdateThrottling(self):
         config_path = self.write_config(SMALL_CLUSTER)
@@ -300,16 +405,22 @@ class AutoscalingTest(unittest.TestCase):
         autoscaler = StandardAutoscaler(
             config_path,
             LoadMetrics(),
+            max_launch_batch=5,
             max_concurrent_launches=5,
             max_failures=0,
             update_interval_s=10)
         autoscaler.update()
-        self.assertEqual(len(self.provider.nodes({})), 2)
+        self.waitForNodes(2)
+        self.assertEqual(autoscaler.num_launches_pending.value, 0)
         new_config = SMALL_CLUSTER.copy()
         new_config["max_workers"] = 1
         self.write_config(new_config)
         autoscaler.update()
-        self.assertEqual(len(self.provider.nodes({})), 2)  # not updated yet
+        # not updated yet
+        # note that node termination happens in the main thread, so
+        # we do not need to add any delay here before checking
+        self.assertEqual(len(self.provider.nodes({})), 2)
+        self.assertEqual(autoscaler.num_launches_pending.value, 0)
 
     def testLaunchConfigChange(self):
         config_path = self.write_config(SMALL_CLUSTER)
@@ -317,18 +428,18 @@ class AutoscalingTest(unittest.TestCase):
         autoscaler = StandardAutoscaler(
             config_path, LoadMetrics(), max_failures=0, update_interval_s=0)
         autoscaler.update()
-        self.assertEqual(len(self.provider.nodes({})), 2)
+        self.waitForNodes(2)
 
         # Update the config to change the node type
         new_config = SMALL_CLUSTER.copy()
         new_config["worker_nodes"]["InstanceType"] = "updated"
         self.write_config(new_config)
-        existing_nodes = set(self.provider.nodes({}))
+        self.provider.ready_to_create.clear()
         for _ in range(5):
             autoscaler.update()
-        new_nodes = set(self.provider.nodes({}))
-        self.assertEqual(len(new_nodes), 2)
-        self.assertEqual(len(new_nodes.intersection(existing_nodes)), 0)
+        self.waitForNodes(0)
+        self.provider.ready_to_create.set()
+        self.waitForNodes(2)
 
     def testIgnoresCorruptedConfig(self):
         config_path = self.write_config(SMALL_CLUSTER)
@@ -336,15 +447,19 @@ class AutoscalingTest(unittest.TestCase):
         autoscaler = StandardAutoscaler(
             config_path,
             LoadMetrics(),
+            max_launch_batch=10,
             max_concurrent_launches=10,
             max_failures=0,
             update_interval_s=0)
         autoscaler.update()
+        self.waitForNodes(2)
 
         # Write a corrupted config
         self.write_config("asdf")
         for _ in range(10):
             autoscaler.update()
+        time.sleep(0.1)
+        self.assertEqual(autoscaler.num_launches_pending.value, 0)
         self.assertEqual(len(self.provider.nodes({})), 2)
 
         # New a good config again
@@ -353,7 +468,7 @@ class AutoscalingTest(unittest.TestCase):
         new_config["max_workers"] = 10
         self.write_config(new_config)
         autoscaler.update()
-        self.assertEqual(len(self.provider.nodes({})), 10)
+        self.waitForNodes(10)
 
     def testMaxFailures(self):
         config_path = self.write_config(SMALL_CLUSTER)
@@ -372,12 +487,12 @@ class AutoscalingTest(unittest.TestCase):
             config_path, LoadMetrics(), max_failures=0, update_interval_s=0)
         autoscaler.update()
         autoscaler.update()
-        self.assertEqual(len(self.provider.nodes({})), 2)
+        self.waitForNodes(2)
         for node in self.provider.mock_nodes.values():
             node.state = "terminated"
         self.assertEqual(len(self.provider.nodes({})), 0)
         autoscaler.update()
-        self.assertEqual(len(self.provider.nodes({})), 2)
+        self.waitForNodes(2)
 
     def testConfiguresNewNodes(self):
         config_path = self.write_config(SMALL_CLUSTER)
@@ -393,7 +508,7 @@ class AutoscalingTest(unittest.TestCase):
             update_interval_s=0)
         autoscaler.update()
         autoscaler.update()
-        self.assertEqual(len(self.provider.nodes({})), 2)
+        self.waitForNodes(2)
         for node in self.provider.mock_nodes.values():
             node.state = "running"
         assert len(
@@ -401,9 +516,7 @@ class AutoscalingTest(unittest.TestCase):
                 TAG_RAY_NODE_STATUS: "uninitialized"
             })) == 2
         autoscaler.update()
-        self.waitFor(
-            lambda: len(self.provider.nodes(
-                {TAG_RAY_NODE_STATUS: "up-to-date"})) == 2)
+        self.waitForNodes(2, tag_filters={TAG_RAY_NODE_STATUS: "up-to-date"})
 
     def testReportsConfigFailures(self):
         config_path = self.write_config(SMALL_CLUSTER)
@@ -419,7 +532,7 @@ class AutoscalingTest(unittest.TestCase):
             update_interval_s=0)
         autoscaler.update()
         autoscaler.update()
-        self.assertEqual(len(self.provider.nodes({})), 2)
+        self.waitForNodes(2)
         for node in self.provider.mock_nodes.values():
             node.state = "running"
         assert len(
@@ -427,9 +540,8 @@ class AutoscalingTest(unittest.TestCase):
                 TAG_RAY_NODE_STATUS: "uninitialized"
             })) == 2
         autoscaler.update()
-        self.waitFor(
-            lambda: len(self.provider.nodes(
-                {TAG_RAY_NODE_STATUS: "update-failed"})) == 2)
+        self.waitForNodes(
+            2, tag_filters={TAG_RAY_NODE_STATUS: "update-failed"})
 
     def testConfiguresOutdatedNodes(self):
         config_path = self.write_config(SMALL_CLUSTER)
@@ -445,13 +557,11 @@ class AutoscalingTest(unittest.TestCase):
             update_interval_s=0)
         autoscaler.update()
         autoscaler.update()
-        self.assertEqual(len(self.provider.nodes({})), 2)
+        self.waitForNodes(2)
         for node in self.provider.mock_nodes.values():
             node.state = "running"
         autoscaler.update()
-        self.waitFor(
-            lambda: len(self.provider.nodes(
-                {TAG_RAY_NODE_STATUS: "up-to-date"})) == 2)
+        self.waitForNodes(2, tag_filters={TAG_RAY_NODE_STATUS: "up-to-date"})
         runner.calls = []
         new_config = SMALL_CLUSTER.copy()
         new_config["worker_setup_commands"] = ["cmdX", "cmdY"]
@@ -472,33 +582,37 @@ class AutoscalingTest(unittest.TestCase):
             config_path, lm, max_failures=0, update_interval_s=0)
         self.assertEqual(len(self.provider.nodes({})), 0)
         autoscaler.update()
-        self.assertEqual(len(self.provider.nodes({})), 2)
+        self.waitForNodes(2)
         autoscaler.update()
+        self.assertEqual(autoscaler.num_launches_pending.value, 0)
         self.assertEqual(len(self.provider.nodes({})), 2)
 
         # Scales up as nodes are reported as used
         lm.update("172.0.0.0", {"CPU": 2}, {"CPU": 0})
         lm.update("172.0.0.1", {"CPU": 2}, {"CPU": 0})
         autoscaler.update()
-        self.assertEqual(len(self.provider.nodes({})), 4)
+        self.waitForNodes(4)
         lm.update("172.0.0.2", {"CPU": 2}, {"CPU": 0})
         autoscaler.update()
-        self.assertEqual(len(self.provider.nodes({})), 6)
+        self.waitForNodes(6)
 
         # Holds steady when load is removed
         lm.update("172.0.0.0", {"CPU": 2}, {"CPU": 2})
         lm.update("172.0.0.1", {"CPU": 2}, {"CPU": 2})
         autoscaler.update()
+        self.assertEqual(autoscaler.num_launches_pending.value, 0)
         self.assertEqual(len(self.provider.nodes({})), 6)
 
         # Scales down as nodes become unused
         lm.last_used_time_by_ip["172.0.0.0"] = 0
         lm.last_used_time_by_ip["172.0.0.1"] = 0
         autoscaler.update()
+        self.assertEqual(autoscaler.num_launches_pending.value, 0)
         self.assertEqual(len(self.provider.nodes({})), 4)
         lm.last_used_time_by_ip["172.0.0.2"] = 0
         lm.last_used_time_by_ip["172.0.0.3"] = 0
         autoscaler.update()
+        self.assertEqual(autoscaler.num_launches_pending.value, 0)
         self.assertEqual(len(self.provider.nodes({})), 2)
 
     def testRecoverUnhealthyWorkers(self):
@@ -515,12 +629,11 @@ class AutoscalingTest(unittest.TestCase):
             node_updater_cls=NodeUpdaterThread,
             update_interval_s=0)
         autoscaler.update()
+        self.waitForNodes(2)
         for node in self.provider.mock_nodes.values():
             node.state = "running"
         autoscaler.update()
-        self.waitFor(
-            lambda: len(self.provider.nodes(
-                {TAG_RAY_NODE_STATUS: "up-to-date"})) == 2)
+        self.waitForNodes(2, tag_filters={TAG_RAY_NODE_STATUS: "up-to-date"})
 
         # Mark a node as unhealthy
         lm.last_heartbeat_time_by_ip["172.0.0.0"] = 0