From 01723d554600d5b35d7125a49342e25960d34c40 Mon Sep 17 00:00:00 2001
From: Joseph Schorr <josephschorr@users.noreply.github.com>
Date: Tue, 22 Dec 2015 15:58:51 -0500
Subject: [PATCH] Catch other cases where the queue item has been removed
Fixes #1096
---
data/queue.py | 36 ++++++++++++++++++++++--------------
1 file changed, 22 insertions(+), 14 deletions(-)
diff --git a/data/queue.py b/data/queue.py
index 3cf921c98..7503e5764 100644
--- a/data/queue.py
+++ b/data/queue.py
@@ -213,23 +213,31 @@ class WorkQueue(object):
def incomplete(self, incomplete_item, retry_after=300, restore_retry=False):
with self._transaction_factory(db):
retry_date = datetime.utcnow() + timedelta(seconds=retry_after)
- incomplete_item_obj = self._item_by_id_for_update(incomplete_item.id)
- incomplete_item_obj.available_after = retry_date
- incomplete_item_obj.available = True
- if restore_retry:
- incomplete_item_obj.retries_remaining += 1
+ try:
+ incomplete_item_obj = self._item_by_id_for_update(incomplete_item.id)
+ incomplete_item_obj.available_after = retry_date
+ incomplete_item_obj.available = True
- incomplete_item_obj.save()
- self._currently_processing = False
- return incomplete_item_obj.retries_remaining > 0
+ if restore_retry:
+ incomplete_item_obj.retries_remaining += 1
+
+ incomplete_item_obj.save()
+ self._currently_processing = False
+ return incomplete_item_obj.retries_remaining > 0
+ except QueueItem.DoesNotExist:
+ return False
def extend_processing(self, item, seconds_from_now, minimum_extension=MINIMUM_EXTENSION):
with self._transaction_factory(db):
- queue_item = self._item_by_id_for_update(item.id)
- new_expiration = datetime.utcnow() + timedelta(seconds=seconds_from_now)
+ try:
+ queue_item = self._item_by_id_for_update(item.id)
+ new_expiration = datetime.utcnow() + timedelta(seconds=seconds_from_now)
+
+ # Only actually write the new expiration to the db if it moves the expiration some minimum
+ if new_expiration - queue_item.processing_expires > minimum_extension:
+ queue_item.processing_expires = new_expiration
+ queue_item.save()
+ except QueueItem.DoesNotExist:
+ return
- # Only actually write the new expiration to the db if it moves the expiration some minimum
- if new_expiration - queue_item.processing_expires > minimum_extension:
- queue_item.processing_expires = new_expiration
- queue_item.save()