From 13ce317a09f9c183fed03267b3f57b16c4a470be Mon Sep 17 00:00:00 2001
From: crusaderky <crusaderky@gmail.com>
Date: Wed, 9 Aug 2023 12:52:52 +0100
Subject: [PATCH] Fix flakiness in tests caused by WindowsTime

---
 distributed/tests/test_scheduler.py |  4 ++--
 distributed/tests/test_spans.py     | 10 +++++-----
 distributed/utils_test.py           | 17 +++++++++++++++++
 3 files changed, 24 insertions(+), 7 deletions(-)

diff --git a/distributed/tests/test_scheduler.py b/distributed/tests/test_scheduler.py
index af476c0915..0bf0652583 100644
--- a/distributed/tests/test_scheduler.py
+++ b/distributed/tests/test_scheduler.py
@@ -2580,7 +2580,7 @@ async def test_no_dangling_asyncio_tasks():
 
 
 @gen_cluster(client=True, Worker=NoSchedulerDelayWorker, config=NO_AMM)
-async def test_task_groups(c, s, a, b):
+async def test_task_groups(c, s, a, b, no_time_resync):
     start = time()
     da = pytest.importorskip("dask.array")
     x = da.arange(100, chunks=(20,))
@@ -2629,7 +2629,7 @@ async def test_task_groups(c, s, a, b):
 
 
 @gen_cluster(client=True, nthreads=[("", 2)], Worker=NoSchedulerDelayWorker)
-async def test_task_groups_update_start_stop(c, s, a):
+async def test_task_groups_update_start_stop(c, s, a, no_time_resync):
     """TaskGroup.stop increases as the tasks in the group finish.
     TaskGroup.start can move backwards in the following use case:
 
diff --git a/distributed/tests/test_spans.py b/distributed/tests/test_spans.py
index 23ca4789d3..28465adac5 100644
--- a/distributed/tests/test_spans.py
+++ b/distributed/tests/test_spans.py
@@ -220,7 +220,7 @@ async def test_no_extension(c, s, a, b):
     Worker=NoSchedulerDelayWorker,
     config={"optimization.fuse.active": False},
 )
-async def test_task_groups(c, s, a, b, release):
+async def test_task_groups(c, s, a, b, release, no_time_resync):
     da = pytest.importorskip("dask.array")
     with span("wf"):
         with span("p1"):
@@ -289,7 +289,7 @@ async def test_task_groups(c, s, a, b, release):
 
 
 @gen_cluster(client=True, nthreads=[("", 1)], Worker=NoSchedulerDelayWorker)
-async def test_before_first_task_finished(c, s, a):
+async def test_before_first_task_finished(c, s, a, no_time_resync):
     t0 = time()
     ev = Event()
     x = c.submit(ev.wait, key="x")
@@ -694,7 +694,7 @@ async def test_active_cpu_seconds_trivial(c, s, a, b):
 
 @pytest.mark.parametrize("some_done", [False, True])
 @gen_cluster(client=True, nthreads=[("", 2)], Worker=NoSchedulerDelayWorker)
-async def test_active_cpu_seconds_not_done(c, s, a, some_done):
+async def test_active_cpu_seconds_not_done(c, s, a, some_done, no_time_resync):
     ev = Event()
     x0 = c.submit(ev.wait, key="x-0", workers=[a.address])
     if some_done:
@@ -722,7 +722,7 @@ async def test_active_cpu_seconds_not_done(c, s, a, some_done):
 
 
 @gen_cluster(client=True, Worker=NoSchedulerDelayWorker)
-async def test_active_cpu_seconds_change_nthreads(c, s, a, b):
+async def test_active_cpu_seconds_change_nthreads(c, s, a, b, no_time_resync):
     ev = Event()
     x = c.submit(ev.wait, key="x", workers=[a.address])
     await wait_for_state("x", "executing", a)
@@ -775,7 +775,7 @@ async def test_active_cpu_seconds_change_nthreads(c, s, a, b):
 
 
 @gen_cluster(client=True, nthreads=[("", 2)], Worker=NoSchedulerDelayWorker)
-async def test_active_cpu_seconds_merged(c, s, a):
+async def test_active_cpu_seconds_merged(c, s, a, no_time_resync):
     """Overlapping input spans are not double-counted
     Empty gap between input spans is not counted
     """
diff --git a/distributed/utils_test.py b/distributed/utils_test.py
index ee2300c7aa..98e77322d4 100644
--- a/distributed/utils_test.py
+++ b/distributed/utils_test.py
@@ -2594,3 +2594,20 @@ def scheduler_delay(self):
     @scheduler_delay.setter
     def scheduler_delay(self, value):
         pass
+
+
+@pytest.fixture()
+def no_time_resync():
+    """Temporarily disable the automatic resync of distributed.metrics._WindowsTime
+    which, every 10 minutes, can cause time() to go backwards a few milliseconds.
+
+    On Linux and MacOSX, this fixture is a no-op.
+    """
+    if WINDOWS:
+        time()  # Initialize or refresh delta
+        bak = time.__self__.next_resync
+        time.__self__.next_resync = float("inf")
+        yield
+        time.__self__.next_resync = bak
+    else:
+        yield