Repository: pydicom/pynetdicom (python)

Issue:
Incoming N-EVENT-REPORT RSP messages during release cause deadlock / exception
**Describe the bug**
We are currently facing an issue when releasing the association. I looks like N-EVENT-REPORT RSP messages are sometimes arriving or sent during the release procedure and cause 3 different kind of behaviours:
- Everything works fine
- We end up in a deadlock in the `while not self._is_paused:` loop inside `release(self)` of `association.py`
- We get the following exception (full log attached):
```
Traceback (most recent call last):
  File "C:\Users\jacob.eckert\AppData\Local\Programs\Python\Python39\lib\threading.py", line 980, in _bootstrap_inner
    self.run()
  File "C:\Users\jacob.eckert\AppData\Local\Programs\Python\Python39\lib\threading.py", line 917, in run
    self._target(*self._args, **self._kwargs)
  File "C:\code\py_env\lib\site-packages\pynetdicom\dul.py", line 445, in run_reactor
    self.state_machine.do_action(event)
  File "C:\code\py_env\lib\site-packages\pynetdicom\fsm.py", line 78, in do_action
    raise InvalidEventError(msg)
pynetdicom.fsm.InvalidEventError: Invalid event 'Evt9' for the current state 'Sta7'
```

Is there a way to ignore that kind of messages (unbinding the handler did not work) or any other clean solution? - Right now we are using a `sleep` before releasing so that no more N-EVENT-REPORT RSP messages are present.

**Expected behavior**
Having a clear and reproducable behaviour on every run

**Steps To Reproduce**
Receive a N-EVENT-REPORT RSP during `assoc.release()` call

**Your environment**

Windows-10-10.0.19044-SP0
Python  3.9.13 (tags/v3.9.13:6de2ca5, May 17 2022, 16:36:42) [MSC v.1929 64 bit (AMD64)]
pydicom  2.3.1
pynetdicom  2.0.2

[full_log.txt](https://github.com/pydicom/pynetdicom/files/10968816/full_log.txt)


Pull request description:
Improvements for handling P-DATA received during association release
<!-- Please use Github's 'Draft Pull Request' for PRs that are in-progress -->
#### Reference issue
* Change FSM to put P-DATA PDUs into the DIMSE queue instead of the `to_user_queue`
* Add `Association._sent_release` flag that is set once association release beings
* Change `Association._serve_request()` to return early if `_sent_release` flag is set
* Add warning to logging if P-DATA received during association release
* Closes #820

#### Tasks
- [x] Unit tests added that reproduce issue or prove feature is working
- [x] Fix or feature added
- [x] Unit tests passing and coverage at 100% after adding fix/feature
- [x] Type annotations updated and passing with mypy


Patch:
diff --git a/docs/changelog/v3.0.0.rst b/docs/changelog/v3.0.0.rst
index a397639cb..3911eab8a 100644
--- a/docs/changelog/v3.0.0.rst
+++ b/docs/changelog/v3.0.0.rst
@@ -16,6 +16,8 @@ Changes
 * Minimum supported pydicom version is 3.0 (:issue:`981`)
 * Added support for Python 3.13
 * Updated SOP classes to version 2025b of the DICOM Standard
+* P-DATA PDUs received during association release are now added to the normal DIMSE
+  messaging queue
 
 Enhancements
 ------------
@@ -24,3 +26,9 @@ Enhancements
   absent from the N-CREATE-RQ can now be done by adding it to the returned *Attribute
   List* dataset (it'll be removed from the dataset prior to sending) (:issue:`995`)
 * Added debugging handlers for the remaining DIMSE-N messages
+
+
+Fixes
+-----
+* Fixed a state machine error caused by receiving N-EVENT-REPORT requests during
+  association release (:issue:`820`)
diff --git a/pynetdicom/acse.py b/pynetdicom/acse.py
index f7dfa02d7..51a7b0306 100644
--- a/pynetdicom/acse.py
+++ b/pynetdicom/acse.py
@@ -282,12 +282,9 @@ class ACSE:
             "a-abort": (A_ABORT,),
             "a-p-abort": (A_P_ABORT,),
         }
-
         primitive = self.dul.peek_next_pdu()
-        if isinstance(primitive, abort_classes[abort_type]):
-            return True
 
-        return False
+        return isinstance(primitive, abort_classes[abort_type])
 
     def is_release_requested(self) -> bool:
         """Return ``True`` if an A-RELEASE request has been received.
@@ -597,14 +594,6 @@ class ACSE:
                 self.assoc.kill()
                 return
 
-            # Any other primitive besides A_RELEASE gets trashed
-            elif not isinstance(primitive, A_RELEASE):
-                # Should only be P-DATA
-                LOGGER.warning(
-                    "P-DATA received after Association release, data has been lost"
-                )
-                continue
-
             # Must be A-RELEASE, but may be either request or release
             if primitive.result is None:
                 # A-RELEASE (request) received, therefore an
@@ -809,6 +798,7 @@ class ACSE:
         if is_response:
             primitive.result = "affirmative"
 
+        self.assoc._sent_release = True
         self.dul.send_pdu(primitive)
 
     def send_request(self) -> None:
diff --git a/pynetdicom/association.py b/pynetdicom/association.py
index 6cf059932..e7f5d3119 100644
--- a/pynetdicom/association.py
+++ b/pynetdicom/association.py
@@ -159,6 +159,8 @@ class Association(threading.Thread):
 
         # Track whether we've sent an abort or not for the abort() method
         self._sent_abort: bool = False
+        # Track whether we've sent a release or not
+        self._sent_release: bool = False
 
         # Accepted and rejected presentation contexts
         self._accepted_cx: dict[int, PresentationContext] = {}
@@ -584,13 +586,14 @@ class Association(threading.Thread):
         return self._rejected_cx
 
     def release(self) -> None:
-        """Initiate association release by send an A-RELEASE request."""
+        """Initiate association release by sending an A-RELEASE request."""
         if self.is_established:
             # Ensure the reactor is paused so it doesn't
             #   steal incoming ACSE messages
             self._reactor_checkpoint.clear()
             while not self._is_paused:
                 time.sleep(0.0001)
+
             LOGGER.info("Releasing Association")
             self.acse.negotiate_release()
             # Restart reactor
@@ -710,7 +713,7 @@ class Association(threading.Thread):
             if msg:
                 self._serve_request(msg, cast(int, context_id))
 
-            # Check for release request
+            # Check for release request from the peer
             if self.acse.is_release_requested():
                 # Send A-RELEASE response
                 self.acse.send_release(is_response=True)
@@ -721,7 +724,7 @@ class Association(threading.Thread):
                 self.kill()
                 return
 
-            # Check for abort
+            # Check for abort from either locally or the peer
             if self.acse.is_aborted():
                 log_msg = "Association Aborted"
                 if self.acse.is_aborted("a-p-abort"):
@@ -751,6 +754,7 @@ class Association(threading.Thread):
                     self._is_paused = False
                 else:
                     self.abort()
+
                 self.kill()
                 return
 
@@ -3515,6 +3519,12 @@ class Association(threading.Thread):
             The ID of the presentation context that the request is being
             made under.
         """
+        if self._sent_release:
+            LOGGER.warning(
+                f"{msg.msg_type} message received during association release, ignoring"
+            )
+            return
+
         # No message or not a service request
         if not msg.is_valid_request:
             LOGGER.warning(f"Received unexpected {msg.msg_type} service message")
diff --git a/pynetdicom/dul.py b/pynetdicom/dul.py
index ff93f5730..c40ced4c0 100644
--- a/pynetdicom/dul.py
+++ b/pynetdicom/dul.py
@@ -38,6 +38,7 @@ if TYPE_CHECKING:  # pragma: no cover
     from pynetdicom.transport import AssociationSocket
 
     _QueueType = queue.Queue[_PDUPrimitiveType | T_CONNECT]
+    _UserQueuePrimitives = A_ASSOCIATE | A_RELEASE | A_ABORT | A_P_ABORT
 
 
 LOGGER = logging.getLogger(__name__)
@@ -83,10 +84,12 @@ class DULServiceProvider(Thread):
         #   user and the DUL service provider.
         # An event occurs when the DUL service user adds to
         #   the to_provider_queue
+        # The queue contains A-ASSOCIATE, A-RELEASE, A-ABORT, A-P-ABORT, P-DATA and
+        #   T-CONNECT primitives from the local user that are to be sent to the peer
         self.to_provider_queue: "_QueueType" = queue.Queue()
         # A primitive is sent to the service user when the DUL service provider
         # adds to the to_user_queue.
-        self.to_user_queue: "queue.Queue[_PDUPrimitiveType]" = queue.Queue()
+        self.to_user_queue: "queue.Queue[_UserQueuePrimitives]" = queue.Queue()
 
         # A queue storing PDUs received from the peer
         self._recv_pdu: "queue.Queue[_PDUType]" = queue.Queue()
@@ -189,10 +192,10 @@ class DULServiceProvider(Thread):
         """Return the network timeout (in seconds)."""
         return self.assoc.network_timeout
 
-    def peek_next_pdu(self) -> "_PDUPrimitiveType | None":
+    def peek_next_pdu(self) -> "_UserQueuePrimitives | None":
         """Check the next PDU to be processed."""
         try:
-            return cast(_PDUPrimitiveType, self.to_user_queue.queue[0])
+            return cast("_UserQueuePrimitives", self.to_user_queue.queue[0])
         except (queue.Empty, IndexError):
             return None
 
@@ -332,7 +335,7 @@ class DULServiceProvider(Thread):
 
     def receive_pdu(
         self, wait: bool = False, timeout: float | None = None
-    ) -> "_PDUPrimitiveType | None":
+    ) -> "_UserQueuePrimitives | None":
         """Return an item from the queue if one is available.
 
         Get the next service primitive to be processed out of the queue of items sent
diff --git a/pynetdicom/fsm.py b/pynetdicom/fsm.py
index f55530d28..16056d87d 100644
--- a/pynetdicom/fsm.py
+++ b/pynetdicom/fsm.py
@@ -503,7 +503,7 @@ def AR_2(dul: "DULServiceProvider") -> str:
         ``'Sta8'``, the next state of the state machine
     """
     # A-RELEASE-RQ PDU received from peer
-    pdu = dul._recv_pdu.get(False)
+    pdu = cast("A_RELEASE_RQ", dul._recv_pdu.get(False))
 
     # Send A-RELEASE indication primitive
     dul.to_user_queue.put(pdu.to_primitive())
@@ -530,7 +530,7 @@ def AR_3(dul: "DULServiceProvider") -> str:
         ``'Sta1'``, the next state of the state machine
     """
     # A-RELEASE-RP PDU received from peer
-    pdu = dul._recv_pdu.get(False)
+    pdu = cast("A_RELEASE_RP", dul._recv_pdu.get(False))
 
     # Issue A-RELEASE confirmation primitive and close transport connection
     dul.to_user_queue.put(pdu.to_primitive())
@@ -628,7 +628,7 @@ def AR_6(dul: "DULServiceProvider") -> str:
     pdu = cast("P_DATA_TF", dul._recv_pdu.get(False))
 
     # Issue P-DATA indication
-    dul.to_user_queue.put(pdu.to_primitive())
+    dul.assoc.dimse.receive_primitive(pdu.to_primitive())
 
     return "Sta7"
 
@@ -762,6 +762,7 @@ def AA_1(dul: "DULServiceProvider") -> str:
     """
     # Received invalid PDU from peer or an A-ABORT primitive from local user
     try:
+        # A-ABORT or A-P-ABORT from local user
         primitive = dul.to_provider_queue.queue[0]
         if isinstance(primitive, (A_ABORT, A_P_ABORT)):
             primitive = dul.to_provider_queue.get(False)


Test patch:
diff --git a/pynetdicom/tests/encoded_pdu_items.py b/pynetdicom/tests/encoded_pdu_items.py
index 48d90302f..00747776b 100644
--- a/pynetdicom/tests/encoded_pdu_items.py
+++ b/pynetdicom/tests/encoded_pdu_items.py
@@ -433,7 +433,7 @@ a_abort = b"\x07\x00\x00\x00\x00\x04\x00\x00\x00\x00"
 a_p_abort = b"\x07\x00\x00\x00\x00\x04\x00\x00\x02\x04"
 
 ################################ P-DATA-TF PDU ###############################
-# Contains a C-ECHO message
+# Contains a C-ECHO-RSP message
 # Context ID: 1
 # Data: \x03\x00\x00\x00\x00\x04\x00\x00\x00\x42\x00\x00\x00\x00\x00\x02\x00
 #       \x12\x00\x00\x00\x31\x2e\x32\x2e\x38\x34\x30\x2e\x31\x30\x30\x30\x38
@@ -453,8 +453,8 @@ a_p_abort = b"\x07\x00\x00\x00\x00\x04\x00\x00\x02\x04"
 #       00 00 02 00 | 12 00 00 00 | 31 2e ... 31 00- Affected SOP Class UID
 #       00 00 00 01 | 02 00 00 00 | 30 80 - Command Field (32816)
 #       00 00 20 01 | 02 00 00 00 | - MessageIDBeingRespondedTo (1)
-#       00 00 00 08 |             | - Command Data Set Type (257)
-#       00 00 00 09 | - Status (0)
+#       00 00 00 08 | 02 00 00 00 | 01 01 - Command Data Set Type (257)
+#       00 00 00 09 | 02 00 00 00 | 00 00 - Status (0)
 p_data_tf = (
     b"\x04\x00\x00\x00\x00\x54\x00\x00\x00\x50\x01\x03\x00\x00\x00"
     b"\x00\x04\x00\x00\x00\x42\x00\x00\x00\x00\x00\x02\x00\x12\x00"
@@ -477,6 +477,31 @@ p_data_tf_rq = (
     b"\x00\x00\x00\x08\x02\x00\x00\x00\x01\x01"  # Command Data Set Type
 )
 
+# N-EVENT-REPORT RQ - Command Set is implicit little
+p_data_tf_n_event_report = (
+    b"\x04\x00\x00\x00\x00\x76"  # P-DATA: 1 type, 2 reserved, 3-6 length
+    b"\x00\x00\x00\x72\x01"  # 7+: PDV Item(s): 1-4 length, 5, CID, 6+ PDV value
+    b"\x03"  # PDV 1, message control header byte
+    # (0000,0000), Length 4, Value 0x64 0x00 0x00 0x000
+    b"\x00\x00\x00\x00\x04\x00\x00\x00\x64\x00\x00\x00"  # Command Group Length
+    # (0000,0002), Length 0x16
+    b"\x00\x00\x02\x00\x16\x00\x00\x00"  # Affected
+    b"\x31\x2e\x32\x2e\x38\x34\x30\x2e\x31\x30\x30\x30\x38\x2e\x35\x2e"
+    b"\x31\x2e\x31\x2e\x31\x34"
+    # (0000,0100), Length 2,
+    b"\x00\x00\x00\x01\x02\x00\x00\x00\x00\x01"
+    # (0000,0110), Length 2
+    b"\x00\x00\x10\x01\x02\x00\x00\x00\x01\x00"
+    # (0000,0800), Length 2
+    b"\x00\x00\x00\x08\x02\x00\x00\x00\x01\x01"  # 0x0101, no data set
+    # (0000,1000), Length 0x116
+    b"\x00\x00\x00\x10\x16\x00\x00\x00\x31\x2e\x32\x2e"
+    b"\x38\x34\x30\x2e\x31\x30\x30\x30\x38\x2e\x35\x2e\x31\x2e\x31\x2e"
+    b"\x31\x37"
+    # (0000,1002), Length 2
+    b"\x00\x00\x02\x10\x02\x00\x00\x00\x03\x00"
+)
+
 # AsynchronousOperationsWindow
 #   Max operations invoked: 5
 #   Max operations performed: 5
diff --git a/pynetdicom/tests/parrot.py b/pynetdicom/tests/parrot.py
index 3e78f63f9..feba65dfe 100644
--- a/pynetdicom/tests/parrot.py
+++ b/pynetdicom/tests/parrot.py
@@ -39,12 +39,16 @@ class ParrotRequest(BaseRequestHandler):
                 self.kill_read = False
                 while not self.kill_read:
                     if self.ready:
-                        self.received.append(bytes(self.read_data))
+                        d = bytes(self.read_data)
+                        self.received.append(d)
                         self.kill_read = True
+                # LOGGER.debug(f"A: Received {len(d)}")
             elif cmd == "send":
+                # LOGGER.debug(f"A: Sending {len(data)}")
                 self.send(data)
                 self.sent.append(data)
             elif cmd == "wait":
+                # LOGGER.debug(f"A: Waiting {data}")
                 time.sleep(data)
 
         # Disconnects automatically when this method ends!
diff --git a/pynetdicom/tests/test_acse.py b/pynetdicom/tests/test_acse.py
index 0573a91d2..50cc6659f 100644
--- a/pynetdicom/tests/test_acse.py
+++ b/pynetdicom/tests/test_acse.py
@@ -49,6 +49,8 @@ from .encoded_pdu_items import (
     a_release_rq,
     a_release_rp,
     p_data_tf,
+    p_data_tf_rq,
+    p_data_tf_n_event_report,
     a_abort,
     a_p_abort,
 )
@@ -2068,33 +2070,6 @@ class TestNegotiateRelease:
         assert scp.received[1] == a_release_rq
         assert scp.received[2] == a_p_abort[:-1] + b"\x00"
 
-    def test_release_p_data(self, caplog):
-        """Test receiving P-DATA-TF after release."""
-        commands = [
-            ("recv", None),
-            ("send", a_associate_ac),
-            ("recv", None),  # a_release_rq
-            ("send", p_data_tf),
-            ("send", a_release_rp),
-        ]
-        self.scp = scp = self.start_server(commands)
-
-        assoc = self.create_assoc()
-        assoc.start()
-        while not assoc.is_established:
-            time.sleep(0.05)
-
-        with caplog.at_level(logging.DEBUG, logger="pynetdicom"):
-            assoc.release()
-            assert assoc.is_released
-            assert (
-                "P-DATA received after Association release, data has been lost"
-            ) in caplog.text
-
-        scp.shutdown()
-
-        assert scp.received[1] == a_release_rq
-
     def test_coll_acc(self, caplog):
         """Test a simulated A-RELEASE collision on the acceptor side."""
 
@@ -2194,6 +2169,63 @@ class TestNegotiateRelease:
         assert scp.received[1] == a_release_rq
         assert scp.received[2] == a_release_rp
 
+    def test_release_n_event_report(self, caplog):
+        """Test P-DATA (N-EVENT-REPORT) received during release."""
+
+        def handle_ner(event):
+            return 0x0000, None
+
+        commands = [
+            ("recv", None),  # a-associate-rq
+            ("send", a_associate_ac),
+            ("recv", None),  # a-release-rq
+            ("send", p_data_tf_n_event_report),  # n-event-report-rq
+            ("wait", 0.5),
+            ("send", a_release_rp),
+        ]
+        self.scp = scp = self.start_server(commands)
+
+        with caplog.at_level(logging.WARNING, logger="pynetdicom"):
+            assoc = self.create_assoc()
+            assoc.bind(evt.EVT_N_EVENT_REPORT, handle_ner)
+            assoc.start()
+            while not assoc.is_established:
+                time.sleep(0.05)
+
+            assoc.acse.send_release(is_response=False)
+
+        scp.shutdown()
+
+        assert (
+            "N-EVENT-REPORT message received during association release, ignoring"
+        ) in caplog.text
+
+    def test_release_c_echo_request(self, caplog):
+        """Test P-DATA (C-ECHO) received during release."""
+        commands = [
+            ("recv", None),  # a-associate-rq
+            ("send", a_associate_ac),
+            ("recv", None),  # a-release-rq
+            ("send", p_data_tf_rq),  # c-echo-rq
+            ("wait", 0.5),
+            ("send", a_release_rp),
+        ]
+        self.scp = scp = self.start_server(commands)
+
+        with caplog.at_level(logging.WARNING, logger="pynetdicom"):
+            assoc = self.create_assoc()
+            assoc.start()
+            while not assoc.is_established:
+                time.sleep(0.05)
+
+            assoc.acse.send_release(is_response=False)
+
+        scp.shutdown()
+
+        assert (
+            "C-ECHO message received during association release, ignoring"
+        ) in caplog.text
+
 
 class TestEventHandlingAcceptor:
     """Test the transport events and handling as acceptor."""