Firestore: watch unsubscribe generates unexpected exception

[rmceoin]

When issuing an unsubscribe() on an existing watch, an exception is emitted. While python continues to function and any new on_snapshot queries work, it's not ideal to have these exceptions.

https://firebase.google.com/docs/firestore/query-data/listen#detach_a_listener

Environment details

1. Specify the API: Firestore
2. OS type and version: Ubuntu 18.04 and CentOS 7 (centos-7-v20190312)
3. Python version and virtual environment information: 3.6.7 (Ubuntu) and 3.6.6 (CentOS)
4. google-cloud- version: pip show google-<service> or pip freeze
   cachetools==3.1.0
   certifi==2019.3.9
   chardet==3.0.4
   google-api-core==1.10.0
   google-auth==1.6.3
   google-cloud-core==0.29.1
   google-cloud-firestore==1.0.0
   googleapis-common-protos==1.5.10
   grpcio==1.20.1
   idna==2.8
   protobuf==3.7.1
   pyasn1==0.4.5
   pyasn1-modules==0.2.5
   pytz==2019.1
   requests==2.21.0
   rsa==4.0
   six==1.12.0
   urllib3==1.24.2

Steps to reproduce

1. Create an on_snapshot watch of a collection.
2. sleep to mimic work being done
3. Invoke an unsubscribe on the watch.
4. Exception errors are emitted to stderr.

Expected behavior is no exception.

Code example

import time
from google.cloud import firestore

def on_snapshot(collection_snapshot, changes, read_time):
    print("on_snapshot()")

client = firestore.Client()
collection_ref = client.collection("localaccts-testing")
watch = collection_ref.on_snapshot(on_snapshot)

while True:
    time.sleep(30)

    watch.unsubscribe()
    watch = collection_ref.on_snapshot(on_snapshot)

Example output showing the exception

on_snapshot()
Thread-ConsumeBidirectionalStream caught unexpected exception <_Rendezvous of RPC that terminated with:
	status = StatusCode.CANCELLED
	details = "Locally cancelled by application!"
	debug_error_string = "None"
> and will exit.
Traceback (most recent call last):
  File "/home/rmceoin/testwatch/env/lib64/python3.6/site-packages/google/api_core/bidi.py", line 543, in _thread_main
    response = self._bidi_rpc.recv()
  File "/home/rmceoin/testwatch/env/lib64/python3.6/site-packages/google/api_core/bidi.py", line 454, in recv
    return self._recoverable(self._recv)
  File "/home/rmceoin/testwatch/env/lib64/python3.6/site-packages/google/api_core/bidi.py", line 413, in _recoverable
    raise exc
  File "/home/rmceoin/testwatch/env/lib64/python3.6/site-packages/google/api_core/bidi.py", line 403, in _recoverable
    return method(*args, **kwargs)
  File "/home/rmceoin/testwatch/env/lib64/python3.6/site-packages/google/api_core/bidi.py", line 451, in _recv
    return next(call)
  File "/home/rmceoin/testwatch/env/lib64/python3.6/site-packages/grpc/_channel.py", line 363, in __next__
    return self._next()
  File "/home/rmceoin/testwatch/env/lib64/python3.6/site-packages/grpc/_channel.py", line 357, in _next
    raise self
grpc._channel._Rendezvous: <_Rendezvous of RPC that terminated with:
	status = StatusCode.CANCELLED
	details = "Locally cancelled by application!"
	debug_error_string = "None"
>
on_snapshot()

[jakeczyz]

We're seeing this too. The errors are logged from a background thread, so they can't be caught/silenced and end up polluting our logs. We see it locally, e.g. when running with:

google-api-core==1.10.0
google-api-python-client==1.7.8
google-cloud-core==0.29.1
google-cloud-error-reporting==0.30.1
google-cloud-firestore==1.0.0
googleapis-common-protos==1.5.10
greenlet==0.4.15
grpc-google-iam-v1==0.11.4
grpcio==1.20.1

as well as when running in a GC Cloud Function Python3.7 environment that just requires the latest google-cloud-firestore. So, I'm pretty sure it's not terribly version/env dependent.

Let us know if more diagnostic data is useful, but this should be trivial to replicate. Thanks!

[tseaver]

@crwilcox ISTM that this should be fixed in google.api_core.bidi: the CANCELLED error should just cause the BIDI stream to shut down gracefully.

[tseaver]

Because the google.api_core.bidi module remains carefully agnostic of specific gRPC status codes (it doesn't even import anything from 'grpc'), I believe the best approach here is to add another argument, should_terminate, to the ResumableBidiRpc class,. Like should_resume. it would take a future / error and return a boolean (true if the stream should terminate). By default, no errors would return true.

Firestore (and maybe Pub/sub) could then customize the set of response codes used to signal "clean" shutdown of the BiDi thread.