dvc-10427 Fix Celery Monitor Shutdowns (#142)

* dvc-10427 Add 2" delay to expirations check

As per treeverse/dvc#10427 investigation we realised that kombu creates expiration dates for messages 1" in the future which affects cleaning the directories on shutdown.

* Add a tip to run a single test to the documentation

* Exclude pyenv virtual environments

* dvc-10427 Send the shutdown message to the appropriate worker.

Please check: treeverse/dvc#10427 (comment)

* Revert "dvc-10427 Add 2" delay to expirations check"

This reverts commit f15acd7 as per treeverse/dvc#10427 (comment)

Author: nablabits

.gitignore

@@ -112,6 +112,7 @@ venv/
 ENV/
 env.bak/
 venv.bak/
+.python-version
 
 # Spyder project settings
 .spyderproject

CONTRIBUTING.rst

@@ -78,6 +78,13 @@ For example, invoke the unit test suite like this:
 
    $ nox --session=tests
 
+During the development process, though, you may be happy to just run the tests you are
+writing in your environment:
+
+.. code:: console
+
+   $ nox --session=tests-3.12 -- -k my_cool_test
+
 Unit tests are located in the ``tests`` directory,
 and are written using the pytest_ testing framework.
 

src/dvc_task/worker/temporary.py

@@ -125,5 +125,5 @@ def _tasksets(nodes):
                 logger.info("monitor: shutting down due to empty queue.")
                 break
         logger.debug("monitor: sending shutdown to '%s'.", nodename)
-        self.app.control.shutdown()
+        self.app.control.shutdown(destination=[nodename])
         logger.debug("monitor: done")

tests/worker/test_temporary.py

@@ -5,6 +5,7 @@
 import pytest
 from celery import Celery
 from celery.concurrency.prefork import TaskPool
+from celery.utils.nodenames import default_nodename
 from celery.worker.worker import WorkController
 from pytest_mock import MockerFixture
 
@@ -52,5 +53,7 @@ def test_monitor(
     """Should shutdown worker when queue empty."""
     worker = TemporaryWorker(celery_app, timeout=1)
     shutdown = mocker.spy(celery_app.control, "shutdown")
-    worker.monitor(celery_worker.hostname)  # type: ignore[attr-defined]
-    shutdown.assert_called_once()
+    hostname = celery_worker.hostname  # type: ignore[attr-defined]
+    worker.monitor(hostname)
+    nodename = default_nodename(hostname)
+    shutdown.assert_called_once_with(destination=[nodename])