Expect and handle exceptions from the AllWatcher task

fixes #829

The `_all_watcher` task is a coroutine for the AllWatcher to run in
the background all the time forever, and it involves a while loop
that's being controlled manually through some flags (asyncio events),
e.g. things like `_watch_stopping`, `watch_stopped`.

The problem is that when the `_all_watcher` raises an exception (or
receives one from things like `get_config()` like in the case of
ether in the event loop, not handled/or re-raised. This is because
this coroutine is not `await`ed (for good reason), it can't be
`await`ed because there won't ever be any results, this method is
supposed to be working in the background forever getting the deltas
for us. As a result of this, if `_all_watcher` fails, then external
flags like `_watch_received` is never set, and whoever's calling
`await self._watch_received.wait()` will block forever (in this case
the `_after_connect()`. Similarly the `disconnect()` waits for the
`_watch_stopped` flag, which won't be set either, so if we call
disconnect when all_watcher failed then it'll hang forever.

This change fixes this problem by allowing (at the wait-for-flag
spots) to wait for two things, 1) whichever flag we're waiting for, 2)
`_all_watcher` task to be `"done()"`. In the latter case, we should
expect to see an exception because that task is not supposed to be
finished. More importantly, if we do see that the
`_watcher_task.done()`, then we don't sit and wait forever for the
_all_watcher event flags to be set, so we won't hang.

Also a nice side effect of this should be that we should be getting
less number of extra exception outputs saying that the "Task exception
is never handled", since we do call the `.exception()` on the
`_all_watcher` task. Though we'll probably continue to get those from
the tasks like `_pinger` and `_debug_log` etc. However, this is a good
first example solution to handle them as well.

Author: cderici

juju/model.py

@@ -759,7 +759,26 @@ async def _after_connect(self):
         # we've received all the model data, which might be
         # a whole load of unneeded data if all the client wants
         # to do is make one RPC call.
-        await self._watch_received.wait()
+        async def watch_received_waiter():
+            await self._watch_received.wait()
+        waiter = jasyncio.create_task(watch_received_waiter())
+
+        # If we just wait for the _watch_received event and the _all_watcher task
+        # fails (e.g. because API fails like migration is in progress), then
+        # we'll hang because the _watch_received will never be set
+        # Instead, we watch for two things, 1) _watch_received, 2) _all_watcher done
+        # If _all_watcher is done before the _watch_received, then we should see
+        # (and raise) an exception coming from the _all_watcher
+        # Otherwise (i.e. _watch_received is set), then we're good to go
+        done, pending = await jasyncio.wait([waiter, self._watcher_task],
+                                            return_when=jasyncio.FIRST_COMPLETED)
+        if self._watcher_task in done:
+            # Cancel the _watch_received.wait
+            waiter.cancel()
+            # If there's no exception, then why did the _all_watcher broke its loop?
+            if not self._watcher_task.exception():
+                raise JujuError("AllWatcher task is finished abruptly without an exception.")
+            raise self._watcher_task.exception()
 
         await self.get_info()
         self.uuid = self.info.uuid
@@ -771,6 +790,12 @@ async def disconnect(self):
         if not self._watch_stopped.is_set():
             log.debug('Stopping watcher task')
             self._watch_stopping.set()
+            # If the _all_watcher task is finished,
+            # check to see an exception, if yes, raise,
+            # otherwise we should see the _watch_stopped
+            # flag is set
+            if self._watcher_task.done() and self._watcher_task.exception():
+                raise self._watcher_task.exception()
             await self._watch_stopped.wait()
             self._watch_stopping.clear()
 
@@ -1040,6 +1065,7 @@ def _watch(self):
         See :meth:`add_observer` to register an onchange callback.
 
         """
+
         def _post_step(obj):
             # Once we get the model, ensure we're running in the correct state
             # as a post step.
@@ -1129,7 +1155,7 @@ async def _all_watcher():
         self._watch_received.clear()
         self._watch_stopping.clear()
         self._watch_stopped.clear()
-        jasyncio.ensure_future(_all_watcher())
+        self._watcher_task = jasyncio.create_task(_all_watcher())
 
     async def _notify_observers(self, delta, old_obj, new_obj):
         """Call observing callbacks, notifying them of a change in model state