From ad9680f7f7b182cd9543165ccae9800de10a8fad Mon Sep 17 00:00:00 2001 From: Ilan Steemers Date: Fri, 13 Nov 2015 21:23:39 +0100 Subject: [PATCH 1/8] Updates botocore and ironmq --- requirements.txt | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/requirements.txt b/requirements.txt index 453c05d..e043bc3 100644 --- a/requirements.txt +++ b/requirements.txt @@ -7,14 +7,14 @@ arrow==0.7.0 blessed==1.14.0 boto3==1.2.1 -botocore==1.3.5 # via boto3 +botocore==1.3.7 # via boto3 django-picklefield==0.3.2 django-redis==4.3.0 docutils==0.12 # via botocore future==0.15.2 hiredis==0.2.0 iron-core==1.1.9 # via iron-mq -iron-mq==0.7 +iron-mq==0.8 jmespath==0.9.0 # via boto3, botocore psutil==3.2.2 pymongo==3.1 From 5a86660dac90669a6e3b27aa96b1930acd902bf3 Mon Sep 17 00:00:00 2001 From: Ilan Steemers Date: Thu, 19 Nov 2015 16:45:22 +0100 Subject: [PATCH 2/8] Updated pymongo for testing --- requirements.txt | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/requirements.txt b/requirements.txt index e043bc3..3e4ec8a 100644 --- a/requirements.txt +++ b/requirements.txt @@ -17,7 +17,7 @@ iron-core==1.1.9 # via iron-mq iron-mq==0.8 jmespath==0.9.0 # via boto3, botocore psutil==3.2.2 -pymongo==3.1 +pymongo==3.1.1 python-dateutil==2.4.2 # via arrow, botocore, iron-core redis==2.10.5 requests==2.8.1 # via iron-core From 3938de84586ee34f892b92824ddaf10876c2afda Mon Sep 17 00:00:00 2001 From: Ilan Steemers Date: Thu, 19 Nov 2015 17:36:26 +0100 Subject: [PATCH 3/8] Downgrades Iron-mq --- requirements.txt | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/requirements.txt b/requirements.txt index 3e4ec8a..533074c 100644 --- a/requirements.txt +++ b/requirements.txt @@ -7,14 +7,14 @@ arrow==0.7.0 blessed==1.14.0 boto3==1.2.1 -botocore==1.3.7 # via boto3 +botocore==1.3.6 # via boto3 django-picklefield==0.3.2 django-redis==4.3.0 docutils==0.12 # via botocore future==0.15.2 hiredis==0.2.0 iron-core==1.1.9 # via iron-mq -iron-mq==0.8 +iron-mq==0.7 jmespath==0.9.0 # via boto3, botocore psutil==3.2.2 pymongo==3.1.1 From aba268de1f18e761f307623849931f97b7a2dc7b Mon Sep 17 00:00:00 2001 From: Ilan Steemers Date: Wed, 20 Jan 2016 11:29:59 +0100 Subject: [PATCH 4/8] Updates botocore for testing --- requirements.txt | 3 +-- 1 file changed, 1 insertion(+), 2 deletions(-) diff --git a/requirements.txt b/requirements.txt index ed6aa46..fafd27f 100644 --- a/requirements.txt +++ b/requirements.txt @@ -7,12 +7,11 @@ arrow==0.7.0 blessed==1.14.1 boto3==1.2.3 -botocore==1.3.20 # via boto3 +botocore==1.3.21 # via boto3 django-picklefield==0.3.2 django-redis==4.3.0 docutils==0.12 # via botocore future==0.15.2 -futures==3.0.4 # via boto3 hiredis==0.2.0 iron-core==1.2.0 # via iron-mq iron-mq==0.8 From c75ab4e7f0c09b0ad1c58ca4cbb56e677c67f074 Mon Sep 17 00:00:00 2001 From: Ilan Steemers Date: Wed, 20 Jan 2016 14:11:53 +0100 Subject: [PATCH 5/8] Only updates existing tasks if it was failing updates stopped, result and success only for existing task results if the original result was failing. Otherwise the result is discarded. --- django_q/cluster.py | 35 ++++++++++++++++++++++------------- 1 file changed, 22 insertions(+), 13 deletions(-) diff --git a/django_q/cluster.py b/django_q/cluster.py index 68d226b..da7c35a 100644 --- a/django_q/cluster.py +++ b/django_q/cluster.py @@ -408,19 +408,28 @@ def save_task(task, broker): try: if task['success'] and 0 < Conf.SAVE_LIMIT <= Success.objects.count(): Success.objects.last().delete() - Task.objects.update_or_create(id=task['id'], - name=task['name'], - defaults={ - 'func': task['func'], - 'hook': task.get('hook'), - 'args': task['args'], - 'kwargs': task['kwargs'], - 'started': task['started'], - 'stopped': task['stopped'], - 'result': task['result'], - 'group': task.get('group'), - 'success': task['success']} - ) + # check if this task has previous results + if Task.objects.filter(id=task['id'], name=task['name']).exists(): + existing_task = Task.objects.get(id=task['id'], name=task['name']) + # only update the result if it hasn't succeeded yet + if not existing_task.success: + existing_task.stopped = task['stopped'] + existing_task.result = task['result'] + existing_task.success = task['success'] + existing_task.save() + else: + Task.objects.create(id=task['id'], + name=task['name'], + func=task['func'], + hook=task.get('hook'), + args=task['args'], + kwargs=task['kwargs'], + started=task['started'], + stopped=task['stopped'], + result=task['result'], + group=task.get('group'), + success=task['success'] + ) except Exception as e: logger.error(e) From 2fbfeaf5eb5d9bc101831603a860a2fcc2818949 Mon Sep 17 00:00:00 2001 From: Ilan Steemers Date: Wed, 20 Jan 2016 14:17:41 +0100 Subject: [PATCH 6/8] docs: updated for new duplicate task handling --- docs/brokers.rst | 7 +++++-- 1 file changed, 5 insertions(+), 2 deletions(-) diff --git a/docs/brokers.rst b/docs/brokers.rst index 1d12c42..60604a2 100644 --- a/docs/brokers.rst +++ b/docs/brokers.rst @@ -18,8 +18,11 @@ Some pointers: * Don't set the :ref:`retry` timer to a lower or equal number than the task timeout. * Retry time includes time the task spends waiting in the clusters internal queue. * Don't set the :ref:`queue_limit` so high that tasks time out while waiting to be processed. -* In case a task is worked on twice, you will see a duplicate key error in the cluster logs. -* Duplicate tasks do generate additional receipt messages, but the result is discarded in favor of the first result. +* In case a task is worked on twice, the task result will be updated with the latest results. +* In some rare cases a non-atomic broker will re-queue a task after it has been acknowledged. +* If a task runs twice and a previous run has succeeded, the new result wil be discarded. +* Limiting the number of retries is handled globally in your actual broker's settings. + Support for more brokers is being worked on. From 72cf1030702c7755496241c1d29a3b460df24344 Mon Sep 17 00:00:00 2001 From: Ilan Steemers Date: Sat, 23 Jan 2016 13:15:06 +0100 Subject: [PATCH 7/8] Updates botocore and psutil for testing --- requirements.txt | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/requirements.txt b/requirements.txt index fafd27f..7cda050 100644 --- a/requirements.txt +++ b/requirements.txt @@ -7,7 +7,7 @@ arrow==0.7.0 blessed==1.14.1 boto3==1.2.3 -botocore==1.3.21 # via boto3 +botocore==1.3.22 # via boto3 django-picklefield==0.3.2 django-redis==4.3.0 docutils==0.12 # via botocore @@ -16,7 +16,7 @@ hiredis==0.2.0 iron-core==1.2.0 # via iron-mq iron-mq==0.8 jmespath==0.9.0 # via boto3, botocore -psutil==3.4.1 +psutil==3.4.2 pymongo==3.2 python-dateutil==2.4.2 # via arrow, botocore, iron-core redis==2.10.5 From 9741efdc5e05d12d564f25c215450690959852ea Mon Sep 17 00:00:00 2001 From: Ilan Steemers Date: Sun, 24 Jan 2016 21:30:42 +0100 Subject: [PATCH 8/8] Adds test for task result update --- django_q/tests/test_cluster.py | 54 +++++++++++++++++++++++++++++++--- 1 file changed, 50 insertions(+), 4 deletions(-) diff --git a/django_q/tests/test_cluster.py b/django_q/tests/test_cluster.py index c5d56eb..fd8fe0d 100644 --- a/django_q/tests/test_cluster.py +++ b/django_q/tests/test_cluster.py @@ -1,16 +1,17 @@ import sys -from multiprocessing import Queue, Event, Value import threading +from multiprocessing import Queue, Event, Value from time import sleep -import os +from django.utils import timezone +import os import pytest myPath = os.path.dirname(os.path.abspath(__file__)) sys.path.insert(0, myPath + '/../') -from django_q.cluster import Cluster, Sentinel, pusher, worker, monitor -from django_q.humanhash import DEFAULT_WORDLIST +from django_q.cluster import Cluster, Sentinel, pusher, worker, monitor, save_task +from django_q.humanhash import DEFAULT_WORDLIST, uuid from django_q.tasks import fetch, fetch_group, async, result, result_group, count_group, delete_group, queue_size from django_q.models import Task, Success from django_q.conf import Conf @@ -340,6 +341,51 @@ def test_bad_secret(broker, monkeypatch): broker.delete_queue() +@pytest.mark.django_db +def test_update_failed(broker): + tag = uuid() + task = {'id': tag[1], + 'name': tag[0], + 'func': 'math.copysign', + 'args': (1, -1), + 'kwargs': {}, + 'started': timezone.now(), + 'stopped': timezone.now(), + 'success': False, + 'result': None} + # initial save - no success + save_task(task, broker) + assert Task.objects.filter(id=task['id']).exists() + saved_task = Task.objects.get(id=task['id']) + assert saved_task.success is False + sleep(0.5) + # second save - no success + old_stopped = task['stopped'] + task['stopped']=timezone.now() + save_task(task, broker) + saved_task = Task.objects.get(id=task['id']) + assert saved_task.stopped > old_stopped + # third save - success + task['stopped']=timezone.now() + task['result']='result' + task['success']=True + save_task(task, broker) + saved_task = Task.objects.get(id=task['id']) + assert saved_task.success is True + # fourth save - no success + task['result'] = None + task['success'] = False + task['stopped'] = old_stopped + save_task(task, broker) + # should not overwrite success + saved_task = Task.objects.get(id=task['id']) + assert saved_task.success is True + assert saved_task.result == 'result' + + + + + @pytest.mark.django_db def assert_result(task): assert task is not None