Increase timeouts for failing tests

distributed/tests/test_resources.py

@@ -249,7 +249,11 @@ async def test_minimum_resource(c, s, a):
     assert a.total_resources == a.available_resources
 
 
-@gen_cluster(client=True, nthreads=[("127.0.0.1", 2, {"resources": {"A": 1}})])
+@gen_cluster(
+    client=True,
+    nthreads=[("127.0.0.1", 2, {"resources": {"A": 1}})],
+    active_rpc_timeout=10,
+)
 async def test_prefer_constrained(c, s, a):
     futures = c.map(slowinc, range(1000), delay=0.1)
     constrained = c.map(inc, range(10), resources={"A": 1})

distributed/tests/test_stress.py

@@ -79,7 +79,7 @@ async def test_cancel_stress(c, s, *workers):
 def test_cancel_stress_sync(loop):
     da = pytest.importorskip("dask.array")
     x = da.random.random((50, 50), chunks=(2, 2))
-    with cluster(active_rpc_timeout=10) as (s, [a, b]):
+    with cluster(active_rpc_timeout=10, disconnect_timeout=10) as (s, [a, b]):
         with Client(s["address"], loop=loop) as c:
             x = c.persist(x)
             y = (x.sum(axis=0) + x.sum(axis=1) + 1).std()

distributed/worker.py

@@ -2521,11 +2521,15 @@ async def execute(self, key, report=False):
                 return
             ts = self.tasks[key]
             if ts.state != "executing":
-                raise RuntimeError(
-                    f"Trying to execute a task {ts} which is not in executing state anymore"
+                # This might happen if keys are canceled
+                logger.debug(
+                    "Trying to execute a task %s which is not in executing state anymore"
+                    % ts
                 )
+                return
             if ts.runspec is None:
-                raise RuntimeError(f"No runspec available for task {ts}")
+                logger.critical("No runspec available for task %s." % ts)
+                return
             if self.validate:
                 assert not ts.waiting_for_data
                 assert ts.state == "executing"