Fix MemoryObjectSendStream.send(falsey) not raising BrokenResourceError when the last receive stream is closed

docs/versionhistory.rst

@@ -12,6 +12,8 @@ This library adheres to `Semantic Versioning 2.0 <http://semver.org/>`_.
 - Added the ``BlockingPortalProvider`` class to aid with constructing synchronous
   counterparts to asynchronous interfaces that would otherwise require multiple blocking
   portals
+- Added ``__slots__`` to ``AsyncResource`` so that child classes can use ``__slots__``
+  (`#733 <https://github.com/agronholm/anyio/pull/733>`_; PR by Justin Su)
 - Fixed erroneous ``RuntimeError: called 'started' twice on the same task status``
   when cancelling a task in a TaskGroup created with the ``start()`` method before
   the first checkpoint is reached after calling ``task_status.started()``
@@ -28,8 +30,10 @@ This library adheres to `Semantic Versioning 2.0 <http://semver.org/>`_.
 - Emit a ``ResourceWarning`` for ``MemoryObjectReceiveStream`` and
   ``MemoryObjectSendStream`` that were garbage collected without being closed (PR by
   Andrey Kazantcev)
-- Added ``__slots__`` to ``AsyncResource`` so that child classes can use ``__slots__``
-  (`#733 <https://github.com/agronholm/anyio/pull/733>`_; PR by Justin Su)
+- Fixed ``MemoryObjectSendStream.send()`` not raising ``BrokenResourceError`` when the
+  last corresponding ``MemoryObjectReceiveStream`` is closed while waiting to send a
+  falsey item (`#731 <https://github.com/agronholm/anyio/issues/731>`_; PR by Ganden
+  Schaffner)
 
 **4.3.0**
 

src/anyio/streams/memory.py

@@ -234,7 +234,8 @@ async def send(self, item: T_contra) -> None:
                 self._state.waiting_senders.pop(send_event, None)
                 raise
 
-            if self._state.waiting_senders.pop(send_event, None):
+            if send_event in self._state.waiting_senders:
+                del self._state.waiting_senders[send_event]
                 raise BrokenResourceError from None
 
     def clone(self) -> MemoryObjectSendStream[T_contra]:

tests/streams/test_memory.py

@@ -224,10 +224,11 @@ async def test_close_send_while_receiving() -> None:
 
 
 async def test_close_receive_while_sending() -> None:
-    send, receive = create_memory_object_stream[str](0)
+    # We send None here as a regression test for #731
+    send, receive = create_memory_object_stream[None](0)
     with pytest.raises(ExceptionGroup) as exc:
         async with create_task_group() as tg:
-            tg.start_soon(send.send, "hello")
+            tg.start_soon(send.send, None)
             await wait_all_tasks_blocked()
             await receive.aclose()