refactor(api): Improve api resource lifecycle management (#19109)

# Overview

This PR contains a few various enhancements that ensure proper resource
garbage collection.

0bd7359 - We never explicitly call
cleanup in Task Queues, which could be an issue if an exception occurs.

0647e68 - There's a closure here that
captures `self`. Removing the closure fixes that.

Author: mjhuff

api/src/opentrons/protocol_api/protocol_context.py

@@ -337,23 +337,24 @@ def clear_commands(self) -> None:
         if self._unsubscribe_commands:
             self._unsubscribe_commands()
 
-        def on_command(message: cmd_types.CommandMessage) -> None:
-            payload = message.get("payload")
+        self._unsubscribe_commands = self.broker.subscribe(
+            cmd_types.COMMAND, self._on_command_callback
+        )
 
-            if payload is None:
-                return
+    def _on_command_callback(self, message: cmd_types.CommandMessage) -> None:
+        """Callback for command messages."""
+        payload = message.get("payload")
 
-            text = payload.get("text")
+        if payload is None:
+            return
 
-            if text is None:
-                return
+        text = payload.get("text")
 
-            if message["$"] == "before":
-                self._commands.append(text)
+        if text is None:
+            return
 
-        self._unsubscribe_commands = self.broker.subscribe(
-            cmd_types.COMMAND, on_command
-        )
+        if message["$"] == "before":
+            self._commands.append(text)
 
     @requires_version(2, 0)
     def is_simulating(self) -> bool:

api/src/opentrons/protocol_runner/task_queue.py

@@ -85,6 +85,11 @@ async def _run(self) -> None:
         except Exception as e:
             log.exception("Exception raised by protocol")
             error = e
+        finally:
+            self._run_func = None
 
-        if self._cleanup_func is not None:
-            await self._cleanup_func(error)
+        try:
+            if self._cleanup_func is not None:
+                await self._cleanup_func(error)
+        finally:
+            self._cleanup_func = None