(fix) Edge case of acquiring lock and immediately cancelled

fifolock.py

@@ -1,3 +1,6 @@
+from asyncio import (
+    CancelledError as _CancelledError,
+)
 from collections import (
     defaultdict as _defaultdict,
     deque as _deque,
@@ -39,7 +42,14 @@ async def __aenter__(self):
         lock_mode = self._lock_mode_type()
         self._waiters.append(lock_mode)
         self._maybe_acquire()
-        await lock_mode
+        try:
+            await lock_mode
+        except _CancelledError:
+            # The waiter could have been resolved, but the task then cancelled
+            if lock_mode.done() and not lock_mode.cancelled():
+                self._holds[type(lock_mode)] -= 1
+            self._maybe_acquire()
+            raise
 
     async def __aexit__(self, _, __, ___):
         self._holds[self._lock_mode_type] -= 1

test.py

@@ -176,6 +176,38 @@ async def test_mutex_cancelled_before_it_acquires_allows_later_mutex(self):
         self.assertEqual(acquisition_history[1], [True, False, False])
         self.assertEqual(acquisition_history[2], [True, False, True])
 
+    @async_test
+    async def test_mutex_cancelled_that_cancels_second_task_allows_third(self):
+
+        lock = FifoLock()
+        done = asyncio.Future()
+        acquired_1 = asyncio.Future()
+        acquired_3 = asyncio.Future()
+
+        async def access_1():
+            try:
+                async with lock(Mutex):
+                    acquired_1.set_result(None)
+                    await done
+            except asyncio.CancelledError:
+                task_2.cancel()
+
+        async def access_2():
+            async with lock(Mutex):
+                await done
+
+        async def access_3():
+            async with lock(Mutex):
+                acquired_3.set_result(None)
+
+        task_1 = asyncio.ensure_future(access_1())
+        task_2 = asyncio.ensure_future(access_2())
+        asyncio.ensure_future(access_3())
+
+        await acquired_1
+        task_1.cancel()
+        await acquired_3
+
     @async_test
     async def test_mutex_requested_concurrently_can_acquire(self):