nostr: handle connection closing while sending initial query

fixes https://console.cloud.google.com/errors/detail/CNzM26W-g4j7lAE;time=P30D?project=granary-demo

README.md

@@ -308,6 +308,8 @@ Changelog
     * Bug fix for when `name` property is an object, eg an `h-card`.
   * `object_to_json`:
     * Convert both `id` and `url` inside `inReplyTo` to `in-reply-to.`
+* `nostr`:
+  * Handle connection closing while sending initial query.
 * `source`:
   * `Source.postprocess`: when extracting @-mentions, defer to existing tag if it has the same `displayName` and has `url`.
 

granary/nostr.py

@@ -538,7 +538,12 @@ def query(self, websocket, filter):
     subscription = secrets.token_urlsafe(16)
     req = ['REQ', subscription, filter]
     logger.debug(f'Sending: {json_dumps(req)}')
-    websocket.send(json_dumps(req))
+
+    try:
+      websocket.send(json_dumps(req))
+    except ConnectionClosedOK as err:
+      logger.warning(err)
+      return []
 
     events = []
     while True:
@@ -559,7 +564,11 @@ def query(self, websocket, filter):
 
     close = ['CLOSE', subscription]
     logger.debug(f'Sending: {json_dumps(close)}')
-    websocket.send(json_dumps(close))
+
+    try:
+      websocket.send(json_dumps(close))
+    except ConnectionClosedOK as err:
+      logger.warning(err)
 
     return events
 
@@ -590,10 +599,10 @@ def create(self, obj, include_link=OMIT_LINK, ignore_formatting=False):
                  ) as websocket:
       create = ['EVENT', event]
       logger.debug(f'Sending: {json_dumps(create)}')
-      websocket.send(json_dumps(create))
       try:
+        websocket.send(json_dumps(create))
         msg = websocket.recv(timeout=HTTP_TIMEOUT)
-      except ConnectionClosed as cc:
+      except ConnectionClosedOK as cc:
         logger.warning(cc)
         return
 

granary/tests/test_nostr.py

@@ -38,14 +38,20 @@
 
 class FakeConnection:
   """Fake of :class:`websockets.sync.client.ClientConnection`."""
-  relay = None
-  sent = []
-  to_receive = []
-  closed = False
-  recv_err = None
+
+  @classmethod
+  def reset(cls):
+    cls.relay = None
+    cls.sent = []
+    cls.to_receive = []
+    cls.closed = False
+    cls.recv_err = cls.send_err = None
 
   @classmethod
   def send(cls, msg):
+    if cls.send_err:
+      raise cls.send_err
+
     assert not cls.closed
     cls.sent.append(json_loads(msg))
     logger.info(msg)
@@ -75,6 +81,10 @@ def fake_connect(uri, open_timeout=None, close_timeout=None):
 
 class NostrTest(testutil.TestCase):
 
+  def setUp(self):
+    super().setUp()
+    FakeConnection.reset()
+
   def test_id_for(self):
     self.assertEqual(
       '9adfa2330b391539f46548ff2e088ea964a2f7374898c7335a86e914cbf2e769',
@@ -467,9 +477,7 @@ def setUp(self):
     self.last_token = 0
     self.mox.stubs.Set(secrets, 'token_urlsafe', self.token)
 
-    FakeConnection.relay = None
-    FakeConnection.sent = []
-    FakeConnection.to_receive = []
+    FakeConnection.reset()
 
     nostr.connect = fake_connect
 
@@ -655,7 +663,16 @@ def test_get_actor_npub(self):
       ['CLOSE', 'towkin 1'],
     ], FakeConnection.sent)
 
-  def test_query_connection_closed_ok_immediate(self):
+  def test_query_connection_closed_ok_send_immediate(self):
+    FakeConnection.send_err = ConnectionClosedOK(None, None)
+
+    with fake_connect('wss://my-relay',
+                      open_timeout=HTTP_TIMEOUT,
+                      close_timeout=HTTP_TIMEOUT,
+                      ) as ws:
+      self.assertEqual([], self.nostr.query(ws, {}))
+
+  def test_query_connection_closed_ok_recv_immediate(self):
     FakeConnection.recv_err = ConnectionClosedOK(None, None)
 
     with fake_connect('wss://my-relay',
@@ -664,7 +681,7 @@ def test_query_connection_closed_ok_immediate(self):
                       ) as ws:
       self.assertEqual([], self.nostr.query(ws, {}))
 
-  def test_query_connection_closed_ok_partial_results(self):
+  def test_query_connection_closed_ok_recv_partial_results(self):
     FakeConnection.to_receive = [
       ['EVENT', 'towkin 1', NOTE_NOSTR],
       ['EVENT', 'towkin 1', NOTE_NOSTR],
@@ -677,7 +694,7 @@ def test_query_connection_closed_ok_partial_results(self):
                       ) as ws:
       self.assertEqual([NOTE_NOSTR, NOTE_NOSTR], self.nostr.query(ws, {}))
 
-  def test_query_connection_closed_error_raises(self):
+  def test_query_connection_closed_error_recv_raises(self):
     FakeConnection.to_receive = [
       ['EVENT', 'towkin 1', NOTE_NOSTR],
     ]