ignore request if protocol closed

lbry/wallet/rpc/session.py

@@ -424,10 +424,11 @@ class RPCSession(SessionBase):
                     self.max_errors = 0
                 self._bump_errors()
             else:
-                for request in requests:
-                    await self._task_group.add(self._handle_request(request))
+                self.schedule_requests(requests)
 
     async def _handle_request(self, request):
+        if self.is_closing():
+            return
         start = time.perf_counter()
         try:
             result = await self.handle_request(request)
@@ -472,6 +473,10 @@ class RPCSession(SessionBase):
     async def handle_request(self, request):
         pass
 
+    def schedule_requests(self, requests):
+        for request in requests:
+            self._task_group.add(self._handle_request(request))
+
     async def send_request(self, method, args=()):
         """Send an RPC request over the network."""
         if self.is_closing():

lbry/wallet/server/session.py

@@ -197,6 +197,8 @@ class SessionManager:
         self.notified_height: typing.Optional[int] = None
         # Cache some idea of room to avoid recounting on each subscription
         self.subs_room = 0
+        self.consumers = 32
+        self.priority_queue = asyncio.PriorityQueue()
 
         self.session_event = Event()
 
@@ -546,7 +548,8 @@ class SessionManager:
             # because we connect to ourself
             await asyncio.wait([
                 self._clear_stale_sessions(),
-                self._manage_servers()
+                self._manage_servers(),
+                self.serve_requests()
             ])
         finally:
             await self._close_servers(list(self.servers.keys()))
@@ -557,6 +560,18 @@ class SessionManager:
                 ])
             await self.stop_other()
 
+    async def serve_requests(self):
+        async def consumer():
+            while True:
+                _, fut = await self.priority_queue.get()
+                try:
+                    await fut
+                except asyncio.CancelledError:
+                    raise
+                except Exception as e:
+                    log.exception("raised while serving a request. This should never happen.")
+        await asyncio.gather(*(consumer() for _ in range(self.consumers)))
+
     async def start_other(self):
         pass
 
@@ -880,6 +895,14 @@ class LBRYElectrumX(SessionBase):
         self.daemon = self.session_mgr.daemon
         self.bp: BlockProcessor = self.session_mgr.bp
         self.db: LevelDB = self.bp.db
+        self.last_request_received_at = 0
+
+    def schedule_requests(self, requests):
+        for request in requests:
+            current = time.perf_counter()
+            elapsed = current - self.last_request_received_at
+            self.last_request_received_at = current
+            self.session_mgr.priority_queue.put_nowait((elapsed, self._handle_request(request)))
 
     @classmethod
     def protocol_min_max_strings(cls):