From a9d6766b5cce3524a9b92f0996e8e3d566483b66 Mon Sep 17 00:00:00 2001 From: Claudio Usai Date: Wed, 29 Apr 2026 11:50:18 +0200 Subject: [PATCH 1/6] calendar-queue: fix _unfinished_tasks accounting in delete_items Each put_nowait increments _unfinished_tasks. delete_items removed items from the heap without ever decrementing the counter, so queue.join() would block forever if any items were deleted before being consumed. Decrement _unfinished_tasks by the number of deleted items (clamped to the current counter value) and set _finished when the count reaches zero, matching the behaviour of task_done(). --- src/calendar_queue/calendar_queue.py | 6 ++++++ 1 file changed, 6 insertions(+) diff --git a/src/calendar_queue/calendar_queue.py b/src/calendar_queue/calendar_queue.py index c1930a9..4ef9ef2 100644 --- a/src/calendar_queue/calendar_queue.py +++ b/src/calendar_queue/calendar_queue.py @@ -258,6 +258,12 @@ def delete_items( # Restore heap invariant after arbitrary deletions. if del_items: heapify(self._queue) + # Mirror the _unfinished_tasks increments done by put_nowait so + # that join() does not block indefinitely for deleted items. + n = min(len(del_items), self._unfinished_tasks) + self._unfinished_tasks -= n + if self._unfinished_tasks == 0: + self._finished.set() self._update_timer() From d25723ff980330487bdcb7b7fed52e34e499cd37 Mon Sep 17 00:00:00 2001 From: Claudio Usai Date: Wed, 29 Apr 2026 11:52:52 +0200 Subject: [PATCH 2/6] calendar-queue: fix unnecessary timer recreation when queue has one item _update_timer guarded the timestamp comparison with 'qsize() > 1', so whenever there was exactly one item in the queue it always cancelled and recreated the timer, even if it was already set for the correct time. The qsize check was never load-bearing: the only thing that matters is whether the existing timer is still pending (when() > loop.time()). Drop the condition so the comparison runs for any queue size. --- src/calendar_queue/calendar_queue.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/src/calendar_queue/calendar_queue.py b/src/calendar_queue/calendar_queue.py index 4ef9ef2..22141e9 100644 --- a/src/calendar_queue/calendar_queue.py +++ b/src/calendar_queue/calendar_queue.py @@ -56,7 +56,7 @@ def _update_timer(self): return prev_timer_ts = self._getter_timer.when() - if self.qsize() > 1 and prev_timer_ts > loop.time(): + if prev_timer_ts > loop.time(): should_update = not math.isclose( prev_timer_ts, loop.time() + ts - time(), abs_tol=0.00001 ) From e8363c55fb553cd724fd792b7796efb080b0703a Mon Sep 17 00:00:00 2001 From: Claudio Usai Date: Wed, 29 Apr 2026 11:49:27 +0200 Subject: [PATCH 3/6] calendar: document `stop()` irreversibility and add `reset()` stop() permanently sets the internal asyncio.Event, causing all future iterations to raise StopAsyncIteration immediately. This was undocumented and left no way to reuse a Calendar instance after stopping. Add reset() which clears the event, re-enabling iteration. --- src/calendar_queue/calendar.py | 18 +++++++++++++++++- 1 file changed, 17 insertions(+), 1 deletion(-) diff --git a/src/calendar_queue/calendar.py b/src/calendar_queue/calendar.py index 50e2ad5..5a5dd1d 100644 --- a/src/calendar_queue/calendar.py +++ b/src/calendar_queue/calendar.py @@ -148,10 +148,26 @@ async def __anext__(self) -> tuple[float, CalendarEvent]: return event def stop(self) -> None: - """Stop the execution of the calendar""" + """Stop the execution of the calendar. + + Once called, any active ``async for`` loop over this instance will + raise ``StopAsyncIteration`` at the next iteration. The effect is + permanent: further iteration will also stop immediately until + ``reset()`` is called. + """ self._stop_event.set() + def reset(self) -> None: + """Re-enable iteration after ``stop()`` has been called. + + Clears the internal stop flag so that the calendar can be iterated + again with ``async for``. Scheduled events that were preserved when + ``stop()`` was called (see ``__anext__``) will be emitted normally. + """ + + self._stop_event.clear() + def clear(self) -> list[tuple[float, CalendarEvent]]: """Cancel all scheduled events. From 559f3359664a93ee2dc8b289f3ee4b79ca14aa68 Mon Sep 17 00:00:00 2001 From: Claudio Usai Date: Wed, 29 Apr 2026 12:05:15 +0200 Subject: [PATCH 4/6] tests: fix fragile `loop.time()` mock in test_far_schedule The previous approach patched loop.time() with a fixed return_value. Because asyncio.wait_for and call_later both rely on loop.time() for their own scheduling, a constant mock breaks those internals. The test only passed because cq.get() happened to short-circuit before any timeout logic ran. Replace the patch with a direct manipulation of the internal timer: cancel the far-future handle and schedule a new one at delay=0, which fires on the next event loop tick. This simulates the item becoming due without touching asyncio's clock. --- tests/test_calendar_queue.py | 24 ++++++++++++++---------- 1 file changed, 14 insertions(+), 10 deletions(-) diff --git a/tests/test_calendar_queue.py b/tests/test_calendar_queue.py index f461210..66a6d45 100644 --- a/tests/test_calendar_queue.py +++ b/tests/test_calendar_queue.py @@ -2,8 +2,6 @@ import sys from datetime import timedelta from time import time -from unittest.mock import patch - import pytest from calendar_queue import CalendarQueue @@ -149,15 +147,12 @@ async def test_delete_restores_heap_property(): @pytest.mark.asyncio async def test_far_schedule(): """Test putting an event scheduled far in time. - We use LoopTimeTravel to simulate the time traveling - and verify that the elements are returned at the correct - timestamp. + Verifies that the item blocks while not yet due, and is returned + once its time arrives. """ cq = CalendarQueue() - cq_loop = cq._get_loop() # type: ignore - delta = timedelta(days=30) cq.put_nowait((time() + delta.total_seconds(), "foo")) @@ -166,9 +161,18 @@ async def test_far_schedule(): await asyncio.wait_for(cq.get(), 2) pytest.fail("Item was returned immediately. It makes no sense.") - with patch.object(cq_loop, "time", return_value=cq_loop.time() + delta.total_seconds() + 10): - item = await asyncio.wait_for(cq.get(), 5) - assert item[-1] == "foo" + # The item is still in the queue (the timed-out get() was cancelled + # before consuming it). Replace the far-future timer with one that + # fires immediately to simulate the scheduled time arriving, without + # globally patching loop.time() which breaks asyncio's own internals. + assert cq._getter_timer is not None # type: ignore + cq._getter_timer.cancel() # type: ignore + cq._getter_timer = cq._get_loop().call_later( # type: ignore + 0, cq._calendar_alarm + ) + + item = await asyncio.wait_for(cq.get(), 5) + assert item[-1] == "foo" @pytest.mark.asyncio From abea2dc3297c9488e11269953660136178b017d3 Mon Sep 17 00:00:00 2001 From: Claudio Usai Date: Wed, 29 Apr 2026 12:26:34 +0200 Subject: [PATCH 5/6] tests: add test for Calendar.reset() Verifies that after stop() is called mid-iteration, further iteration is immediately blocked, and that reset() re-enables the async iterator so remaining scheduled events are emitted correctly. --- tests/test_calendar.py | 34 ++++++++++++++++++++++++++++++++++ 1 file changed, 34 insertions(+) diff --git a/tests/test_calendar.py b/tests/test_calendar.py index 2133d8f..cb9350d 100644 --- a/tests/test_calendar.py +++ b/tests/test_calendar.py @@ -77,6 +77,40 @@ async def test_events_generator(): c.stop() +@pytest.mark.asyncio +async def test_reset(): + """Test that reset() re-enables iteration after stop() has been called.""" + + c: Calendar[int] = Calendar() + + ts = datetime.now() + + for i in range(4): + c.schedule(i, ts + timedelta(seconds=i * 1)) + + # Consume the first two events then stop. + collected = [] + async for _, event in c: + collected.append(event) + if event == 1: + c.stop() + + assert collected == [0, 1] + + # Without reset, iteration stops immediately. + async for _, event in c: + pytest.fail(f"Should not have received event {event} after stop()") + + # After reset, the remaining events are emitted normally. + c.reset() + async for _, event in c: + collected.append(event) + if len(c.events()) == 0: + c.stop() + + assert collected == [0, 1, 2, 3] + + @pytest.mark.asyncio async def test_cancel_events(): """Test that cancelling events remove them from the calendar From 06b50ef0ad65cd4e1a0b0c502df3571139e59b3c Mon Sep 17 00:00:00 2001 From: Claudio Usai Date: Wed, 29 Apr 2026 12:15:25 +0200 Subject: [PATCH 6/6] tests: increase too strict tolerance causing CI failures Signed-off-by: Claudio Usai --- tests/__init__.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/tests/__init__.py b/tests/__init__.py index 6c12b18..96484e8 100644 --- a/tests/__init__.py +++ b/tests/__init__.py @@ -1 +1 @@ -ABS_TOLERANCE = 0.0001 +ABS_TOLERANCE = 0.001