PubSub: Policy.on_exception actually used to make consumer go inactive. - #4472

OK I have updated the gist to be much clearer as to how things were behaving and how they behave now. Most notable:

• Heartbeats where running=True, done=True in the logs_pypi.txt (happens immediately after thread.Policy.on_exception is called)
• Lease management (e.g. "Snoozing lease management...") ceases in logs_local.txt immediately after the heartbeats change to running=False, done=True. Just to be sure, we have 40 seconds worth of heartbeats before stopping the main thread and there is still no lease management happening

Remaining point of discussion (got buried below the fold):

... AFAICT thread.Policy.close() (which gets called by BasePolicy.lease()) is what cleans up (it's the only thing that can stop Consumer._blocking_consume since Consumer.stop_consuming is the only thing that calls self._exiting.set()).

OK my previous remark seems to be wrong, instead it seems that BasePolicy.maintain_leases() is where the action is.

I am going to merge this since it "moves the bar" towards correctness even though I know it isn't 100% correct. E.g. I can still trigger "set_exception can only be called once." by using a subscription that doesn't exist.