tornado: Handle the handler having been cleared by connection close.

As premonitioned in c741c527d7, it is
indeed possible for `get_handler_by_id` to error out by cause the
handler has been unset elsewhere.

Protect the callsites of `get_handler_by_id` to be able to gracefully
handle when the handler has already done away.

zerver/tornado/event_queue.py

@@ -197,6 +197,7 @@ class ClientDescriptor:
     def add_event(self, event: Mapping[str, Any]) -> None:
         if self.current_handler_id is not None:
             handler = get_handler_by_id(self.current_handler_id)
+            if handler is not None:
                 assert handler._request is not None
                 async_request_timer_restart(handler._request)
 

zerver/tornado/handlers.py

@@ -24,8 +24,8 @@ handlers: Dict[int, "AsyncDjangoHandler"] = {}
 fake_wsgi_container = WSGIContainer(lambda environ, start_response: [])
 
 
-def get_handler_by_id(handler_id: int) -> "AsyncDjangoHandler":
-    return handlers[handler_id]
+def get_handler_by_id(handler_id: int) -> Optional["AsyncDjangoHandler"]:
+    return handlers.get(handler_id)
 
 
 def allocate_handler_id(handler: "AsyncDjangoHandler") -> int:
@@ -52,12 +52,18 @@ def finish_handler(handler_id: int, event_queue_id: str, contents: List[Dict[str
         from zerver.lib.request import RequestNotes
         from zerver.middleware import async_request_timer_restart
 
+        # The request handler may have been GC'd by a
+        # on_connection_close elsewhere already, so we have to check
+        # it is still around.
+        handler = get_handler_by_id(handler_id)
+        if handler is None:
+            return
+        request = handler._request
+        assert request is not None
+
         # We call async_request_timer_restart here in case we are
         # being finished without any events (because another
         # get_events request has supplanted this request)
-        handler = get_handler_by_id(handler_id)
-        request = handler._request
-        assert request is not None
         async_request_timer_restart(request)
         log_data = RequestNotes.get_notes(request).log_data
         assert log_data is not None