Merge pull request #98 from backtick-se/stability_fixes

Await orphaned tasks, task failure tests & agent error storage fix

.flake8

@@ -0,0 +1,8 @@
+[flake8]
+max-line-length = 100
+exclude =
+    __pycache__,
+    .git,
+    build,
+    cloud,
+    dist

cowait/engine/test_docker.py

@@ -116,3 +116,36 @@ def test_docker_child_task():
 
     children = dp.find_child_containers(task.id)
     assert len(children) == 0
+
+
+def test_docker_task_error():
+    dp = DockerProvider()
+
+    task = dp.spawn(TaskDefinition(
+        name=TEST_TASK,
+        image=TEST_IMAGE,
+        inputs={'error': True},
+    ))
+
+    container = dp.docker.containers.get(task.container.id)
+    assert task.container == container
+
+    result = container.wait()
+    assert result['StatusCode'] != 0
+
+
+def test_docker_child_error():
+    dp = DockerProvider()
+
+    task = dp.spawn(TaskDefinition(
+        name=TEST_TASK,
+        image=TEST_IMAGE,
+        inputs={'child_error': True},
+    ))
+
+    container = dp.docker.containers.get(task.container.id)
+    assert task.container == container
+
+    # child error should cause the parent to fail
+    result = container.wait()
+    assert result['StatusCode'] != 0

cowait/network/client.py

@@ -45,7 +45,7 @@ async def _connect(self, url: str, token: str) -> None:
 
                 # send buffered messages
                 for msg in self.buffer:
-                    self.send(msg)
+                    await self.send(msg)
                 self.buffer = []
 
                 # client loop

cowait/tasks/remote_task.py

@@ -22,6 +22,10 @@ def __init__(self, taskdef: TaskDefinition, cluster):
     def __await__(self):
         return self.awaitable.__await__()
 
+    @property
+    def done(self):
+        return self.future.done()
+
     def destroy(self):
         self.cluster.destroy(self.id)
 

cowait/test/tasks/utility_task.py

@@ -13,17 +13,28 @@ def __init__(self, taskdef, cluster, node):
         # store taskdef for later comparison
         self.taskdef = taskdef
 
-    async def run(self, child: bool = False, **inputs):
+    async def run(
+        self,
+        child: bool = False,
+        error: bool = False,
+        child_error: bool = False,
+        **inputs
+    ):
         # dump task information to stdout
         print(json.dumps({
             'taskdef': self.taskdef.serialize(),
             'env': dict(os.environ),
         }))
 
         # create a child
-        if child:
+        if child or child_error:
             print('spawn child')
-            await self.spawn('cowait.test.tasks.utility_task', inputs={'child': False})
+            await self.spawn(
+                'cowait.test.tasks.utility_task',
+                inputs={'error': child_error})
+
+        if error:
+            raise RuntimeError('Caused test error')
 
         # run forever
         if inputs.get('forever', False):

cowait/utils/emitter.py

@@ -5,7 +5,7 @@ def __init__(self):
 
     def on(self, type: str, callback: callable) -> None:
         try:
-            self.callbacks[type].append(callback)
+            self.callbacks[type].insert(0, callback)
         except KeyError:
             self.callbacks[type] = [callback]
 

cowait/worker/executor.py

@@ -70,6 +70,12 @@ async def execute(cluster: ClusterProvider, taskdef: TaskDefinition) -> None:
             # execute task
             result = await task.run(**inputs)
 
+            # wait for dangling tasks
+            orphans = filter(lambda child: not child.done, task.subtasks.values())
+            for orphan in orphans:
+                print('~~ waiting for orphaned task', orphan.id)
+                await orphan
+
             # after hook
             await task.after(inputs)
 
@@ -86,8 +92,9 @@ async def execute(cluster: ClusterProvider, taskdef: TaskDefinition) -> None:
     except TaskError as e:
         # pass subtask errors upstream
         await node.parent.send_fail(
-            f'Caught exception in {taskdef.id}:\n'
+            f'Caught exception in subtask {taskdef.id}:\n'
             f'{e.error}')
+        raise e
 
     except Exception as e:
         # capture local errors