Merge pull request kragniz#1109 from cenkalti/lock

Fix for missing DELETE events while waiting for lock

Author: kragniz

.dockerignore

@@ -23,6 +23,7 @@ var/
 *.egg-info/
 .installed.cfg
 *.egg
+.direnv/
 
 # PyInstaller
 #  Usually these files are written by a python script from a template

Dockerfile

@@ -13,7 +13,8 @@ RUN pip install -U tox
 RUN mkdir python-etcd3
 WORKDIR python-etcd3
 # Rebuild this layer .tox when tox.ini or requirements changes
-COPY tox.ini requirements.txt test-requirements.txt ./
+COPY tox.ini ./
+COPY requirements/base.txt requirements/test.txt ./requirements/
 
 RUN tox -epy35 --notest
 

etcd3/locks.py

@@ -1,8 +1,8 @@
+import threading
+import time
 import uuid
 
-import tenacity
-
-from etcd3 import exceptions
+from etcd3 import events, exceptions
 
 
 class Lock(object):
@@ -55,54 +55,64 @@ def acquire(self, timeout=10):
         :returns: True if the lock has been acquired, False otherwise.
 
         """
-        stop = (
-            tenacity.stop_never
-            if timeout is None else tenacity.stop_after_delay(timeout)
-        )
+        if timeout is not None:
+            deadline = time.time() + timeout
 
-        def wait(previous_attempt_number, delay_since_first_attempt):
-            if timeout is None:
-                remaining_timeout = None
-            else:
-                remaining_timeout = max(timeout - delay_since_first_attempt, 0)
-            # TODO(jd): Wait for a DELETE event to happen: that'd mean the lock
-            # has been released, rather than retrying on PUT events too
-            try:
-                self.etcd_client.watch_once(self.key, remaining_timeout)
-            except exceptions.WatchTimedOut:
-                pass
-            return 0
-
-        @tenacity.retry(retry=tenacity.retry_never,
-                        stop=stop,
-                        wait=wait)
-        def _acquire():
-            # TODO: save the created revision so we can check it later to make
-            # sure we still have the lock
-
-            self.lease = self.etcd_client.lease(self.ttl)
-
-            success, _ = self.etcd_client.transaction(
-                compare=[
-                    self.etcd_client.transactions.create(self.key) == 0
-                ],
-                success=[
-                    self.etcd_client.transactions.put(self.key, self.uuid,
-                                                      lease=self.lease)
-                ],
-                failure=[
-                    self.etcd_client.transactions.get(self.key)
-                ]
-            )
-            if success is True:
+        while True:
+            if self._try_acquire():
                 return True
-            self.lease = None
-            raise tenacity.TryAgain
 
+            if timeout is not None:
+                remaining_timeout = max(deadline - time.time(), 0)
+                if remaining_timeout == 0:
+                    return False
+            else:
+                remaining_timeout = None
+
+            self._wait_delete_event(remaining_timeout)
+
+    def _try_acquire(self):
+        self.lease = self.etcd_client.lease(self.ttl)
+
+        success, metadata = self.etcd_client.transaction(
+            compare=[
+                self.etcd_client.transactions.create(self.key) == 0
+            ],
+            success=[
+                self.etcd_client.transactions.put(self.key, self.uuid,
+                                                  lease=self.lease)
+            ],
+            failure=[
+                self.etcd_client.transactions.get(self.key)
+            ]
+        )
+        if success is True:
+            self.revision = metadata[0].response_put.header.revision
+            return True
+        self.revision = metadata[0][0][1].mod_revision
+        self.lease = None
+        return False
+
+    def _wait_delete_event(self, timeout):
         try:
-            return _acquire()
-        except tenacity.RetryError:
-            return False
+            event_iter, cancel = self.etcd_client.watch(
+                self.key, start_revision=self.revision + 1)
+        except exceptions.WatchTimedOut:
+            return
+
+        if timeout is not None:
+            timer = threading.Timer(timeout, cancel)
+            timer.start()
+        else:
+            timer = None
+
+        for event in event_iter:
+            if isinstance(event, events.DeleteEvent):
+                if timer is not None:
+                    timer.cancel()
+
+                cancel()
+                break
 
     def release(self):
         """Release the lock."""

requirements/base.in

@@ -1,3 +1,2 @@
 grpcio>=1.2.0
-tenacity==5.0.2
-protobuf==3.6.1
+protobuf==3.6.1

requirements/base.txt

@@ -1,14 +1,15 @@
-# SHA1:c7df483d2d529decbe2ba01bd60823437a5199a3
+# SHA1:54fb4872dad11196602da52f5e2fb464827b7da8
 #
 # This file is autogenerated by pip-compile-multi
 # To update, run:
 #
 #    pip-compile-multi
 #
-grpcio==1.27.1
-protobuf==3.6.1
-six==1.12.0               # via grpcio, protobuf, tenacity
-tenacity==6.1.0
+enum34==1.1.10            # via grpcio
+futures==3.3.0            # via grpcio
+grpcio==1.28.1            # via -r base.in
+protobuf==3.6.1           # via -r base.in
+six==1.14.0               # via grpcio, protobuf
 
 # The following packages are considered to be unsafe in a requirements file:
-# setuptools==41.2.0        # via protobuf
+# setuptools

requirements/test.in

@@ -15,3 +15,4 @@ flake8-docstrings==1.3.0
 pydocstyle<4  # python2.7
 mock==2.0.0
 pifpaf
+tenacity==5.0.2

requirements/test.txt

@@ -1,4 +1,4 @@
-# SHA1:6bbf20a51a6aeaf6f5aa1d3bb0f3bedfbdee7807
+# SHA1:ed7b4404625d5a47343a116a02b2f8baa7ceb80b
 #
 # This file is autogenerated by pip-compile-multi
 # To update, run:
@@ -7,68 +7,83 @@
 #
 -r base.txt
 alabaster==0.7.12         # via sphinx
+appdirs==1.4.3            # via virtualenv
 argparse==1.4.0           # via unittest2
 atomicwrites==1.3.0       # via pytest
-attrs==19.2.0             # via hypothesis, pytest
+attrs==19.3.0             # via hypothesis, pytest
 babel==2.8.0              # via sphinx
-bumpversion==0.5.3
-certifi==2019.9.11        # via requests
-cffi==1.12.3              # via xattr
+bumpversion==0.5.3        # via -r test.in
+certifi==2020.4.5.1       # via requests
+cffi==1.14.0              # via xattr
 chardet==3.0.4            # via requests
-click==7.0                # via pifpaf
-coverage==5.0.1
+click==7.1.1              # via pifpaf
+configparser==4.0.2       # via entrypoints, flake8, importlib-metadata, pydocstyle
+contextlib2==0.6.0.post1  # via importlib-metadata, importlib-resources, virtualenv, zipp
+coverage==5.0.4           # via -r test.in, pytest-cov
 daiquiri==2.1.1           # via pifpaf
-docutils==0.15.2          # via sphinx
+distlib==0.3.0            # via virtualenv
+docutils==0.16            # via sphinx
 entrypoints==0.3          # via flake8
 extras==1.0.0             # via testtools
-filelock==3.0.12          # via tox
+filelock==3.0.12          # via tox, virtualenv
 fixtures==3.0.0           # via pifpaf, testtools
-flake8-docstrings==1.3.0
-flake8-import-order==0.18.1
+flake8-docstrings==1.3.0  # via -r test.in
+flake8-import-order==0.18.1  # via -r test.in
 flake8-polyfill==1.0.2    # via flake8-docstrings
-flake8==3.7.8
-grpcio-tools==1.27.2
-hypothesis==4.56.3
-idna==2.8                 # via requests
-imagesize==1.1.0          # via sphinx
-importlib-metadata==1.5.0  # via pluggy, pytest
+flake8==3.7.9             # via -r test.in, flake8-docstrings, flake8-polyfill
+funcsigs==1.0.2           # via mock, pytest
+functools32==3.2.3.post2  # via flake8
+grpcio-tools==1.28.1      # via -r test.in
+hypothesis==4.57.1        # via -r test.in
+idna==2.9                 # via requests
+imagesize==1.2.0          # via sphinx
+importlib-metadata==1.6.0  # via importlib-resources, pluggy, pytest, virtualenv
+importlib-resources==1.4.0  # via virtualenv
 jinja2==2.11.1            # via pifpaf, sphinx
 linecache2==1.0.0         # via traceback2
 markupsafe==1.1.1         # via jinja2
 mccabe==0.6.1             # via flake8
-mock==2.0.0
-more-itertools==5.0.0
+mock==2.0.0               # via -r test.in
+monotonic==1.5            # via tenacity
+more-itertools==5.0.0     # via -r test.in, pytest
 packaging==20.3           # via pytest, sphinx
-pbr==5.4.3                # via fixtures, mock, pifpaf, testtools
-pifpaf==2.2.2
-pluggy==0.13.0            # via pytest, tox
-psutil==5.6.3             # via pifpaf
-py==1.8.0                 # via pytest, tox
+pathlib2==2.3.5           # via importlib-metadata, importlib-resources, pytest, virtualenv
+pbr==5.4.5                # via fixtures, mock, pifpaf, testtools
+pifpaf==2.4.0             # via -r test.in
+pluggy==0.13.1            # via pytest, tox
+psutil==5.7.0             # via pifpaf
+py==1.8.1                 # via pytest, tox
 pycodestyle==2.5.0        # via flake8, flake8-import-order
 pycparser==2.20           # via cffi
-pydocstyle==3.0.0
+pydocstyle==3.0.0         # via -r test.in, flake8-docstrings
 pyflakes==2.1.1           # via flake8
-pygments==2.4.2           # via sphinx
-pyparsing==2.4.6          # via packaging
-pytest-cov==2.7.1
-pytest==4.6.5
+pygments==2.5.2           # via sphinx
+pyparsing==2.4.7          # via packaging
+pytest-cov==2.8.1         # via -r test.in
+pytest==4.6.5             # via -r test.in, pytest-cov
+python-json-logger==0.1.11  # via daiquiri
 python-mimeparse==1.6.0   # via testtools
-pytz==2019.2              # via babel
-pyyaml==5.3.1
-requests==2.22.0          # via sphinx
+pytz==2019.3              # via babel
+pyyaml==5.1               # via -r test.in
+requests==2.23.0          # via sphinx
+scandir==1.10.0           # via pathlib2
+singledispatch==3.4.0.3   # via importlib-resources
 snowballstemmer==2.0.0    # via pydocstyle, sphinx
-sphinx==1.8.2
+sortedcontainers==2.1.0   # via hypothesis
+sphinx==1.8.2             # via -r test.in
 sphinxcontrib-websupport==1.1.2  # via sphinx
+tenacity==5.0.2           # via -r test.in
 testtools==2.4.0          # via fixtures
 toml==0.10.0              # via tox
-tox==3.14.5
+tox==3.5.3                # via -r test.in
 traceback2==1.4.0         # via testtools, unittest2
+typing==3.7.4.1           # via flake8, importlib-resources, sphinx
 unittest2==1.1.0          # via testtools
 urllib3==1.25.8           # via requests
-virtualenv==20.0.13        # via tox
+virtualenv==20.0.17       # via tox
 wcwidth==0.1.9            # via pytest
-xattr==0.9.6              # via pifpaf
-zipp==1.0.0               # via importlib-metadata
+xattr==0.9.7              # via pifpaf
+zipp==1.2.0               # via importlib-metadata, importlib-resources
 
 # The following packages are considered to be unsafe in a requirements file:
-# setuptools==41.2.0        # via flake8-import-order, protobuf, sphinx, tox
+# setuptools

tests/test_etcd3.py

@@ -843,6 +843,40 @@ def test_lock_acquire_none(self, etcd):
         # lock is not refreshed
         assert lock.acquire(None) is True
 
+    def test_lock_acquire_with_timeout(self, etcd):
+        lock1 = etcd.lock('lock-10', ttl=10)
+        lock2 = etcd.lock('lock-10', ttl=10)
+
+        original_watch = etcd.watch
+        watch_called = [0]
+
+        def release_lock_before_watch(*args, **kwargs):
+            watch_called[0] += 1
+            # Simulates the case where key is expired before watch is called.
+            # See https://github.com/kragniz/python-etcd3/issues/1107
+            lock1.release()
+            return original_watch(*args, **kwargs)
+
+        original_transaction = etcd.transaction
+        transaction_called = [0]
+
+        def transaction_wrapper(*args, **kwargs):
+            transaction_called[0] += 1
+            return original_transaction(*args, **kwargs)
+
+        assert lock1.acquire() is True
+        with mock.patch.object(etcd3.Etcd3Client, 'watch',
+                               wraps=release_lock_before_watch):
+            with mock.patch.object(etcd3.Etcd3Client, 'transaction',
+                                   wraps=transaction_wrapper):
+                assert lock2.acquire(timeout=5) is True
+
+        # watch must be called only for lock2 once
+        assert watch_called[0] == 1
+
+        # transaction must be called once for lock1, twice for lock2
+        assert transaction_called[0] == 3
+
     def test_internal_exception_on_internal_error(self, etcd):
         exception = self.MockedException(grpc.StatusCode.INTERNAL)
         kv_mock = mock.MagicMock()