fix: Watch thread deadlock on exit (#1014)

* fix: fix thread cleanup when destroying Watch instance

* added test for on_snapshot while closing

Author: daniel-sanche

packages/google-cloud-firestore/google/cloud/firestore_v1/watch.py

@@ -440,6 +440,9 @@ def on_snapshot(self, proto):
             proto(`google.cloud.firestore_v1.types.ListenResponse`):
                 Callback method that receives a object to
         """
+        if self._closing.locked():
+            # don't process on_snapshot responses while spinning down, to prevent deadlock
+            return
         if proto is None:
             self.close()
             return

packages/google-cloud-firestore/tests/unit/v1/test_watch.py

@@ -400,6 +400,15 @@ def test_watch_on_snapshot_target_w_none():
     assert inst._rpc is None
 
 
+def test_watch_on_snapshot_while_closing():
+    inst = _make_watch()
+    inst.close = mock.Mock()
+    with inst._closing:
+        inst.on_snapshot(mock.Mock())
+        # close should not be called again when already closing
+        inst.close.assert_not_called()
+
+
 def test_watch_on_snapshot_target_no_change_no_target_ids_not_current():
     inst = _make_watch()
     proto = _make_listen_response()