From 535a14f4ea1973eaabcaf4624604a42335b15981 Mon Sep 17 00:00:00 2001 From: schneefux Date: Thu, 9 Mar 2017 18:45:28 +0100 Subject: allow non critical job fails --- worker.py | 35 +++++++++++++++++++++-------------- 1 file changed, 21 insertions(+), 14 deletions(-) (limited to 'worker.py') diff --git a/worker.py b/worker.py index 2115847..2d33537 100644 --- a/worker.py +++ b/worker.py @@ -48,21 +48,28 @@ class Worker(object): continue await self._windup() - error = None - try: - for jobid, payload, priority in jobs: + critical_error = True + failed = [] + for jobid, payload, priority in jobs: + try: await self._execute_job(jobid, payload, priority) - except JobFailed as err: - error = err.args[0] - finally: - if error is not None: - await self._queue.reset([j[0] for j in jobs]) - await self._queue.fail(jobid, error) - logging.warning("batch failed, reset") - await self._teardown(failed=True) - else: - await self._queue.finish([j[0] for j in jobs]) - await self._teardown(failed=False) + except JobFailed as err: + failed.append((jobid, err.args[0])) + if len(err.args) > 1: + # rollback + critical_error = err.args[1] + if critical_error: + break + if critical_error: + await self._queue.reset([j[0] for j in jobs]) + logging.warning("batch failed, reset") + else: + await self._queue.finish([j[0] for j in jobs]) + + for jobid, err in failed: + await self._queue.fail(jobid, err) + + await self._teardown(failed=critical_error) async def start(self, batchlimit=1): """Start in background.""" -- cgit v1.3.1