Что случилось с этой настройкой очереди задач?

Question

Я настроил реализацию очереди задач на сайте, который я размещаю для клиента, у него есть cron job, который запускается каждое утро в 2 часа ночи «/admin/tasks/queue», это отправляет электронные письма для отправки «/admin/tasks/email» и использует cursors, поэтому как делать очереди в небольших кусках. По какой-то причине вчера вечером /admin/tasks/queue продолжал получать этот код и поэтому отправил всю свою квоту по электронной почте :/. Я сделал что-то не так с этим кодом?

class QueueUpEmail(webapp.RequestHandler): 
    def post(self): 
     subscribers = Subscriber.all() 
     subscribers.filter("verified =", True) 

     last_cursor = memcache.get('daily_email_cursor') 
     if last_cursor: 
      subscribers.with_cursor(last_cursor) 

     subs = subscribers.fetch(10) 
     logging.debug("POST - subs count = %i" % len(subs)) 
     if len(subs) < 10: 
      logging.debug("POST - Less than 10 subscribers in subs") 
      # Subscribers left is less than 10, don't reschedule the task 
      for sub in subs: 
       task = taskqueue.Task(url='/admin/tasks/email', params={'email': sub.emailaddress, 'day': sub.day_no}) 
       task.add("email") 
      memcache.delete('daily_email_cursor') 
     else: 
      logging.debug("POST - Greater than 10 subscibers left in subs - reschedule") 
      # Subscribers is 10 or greater, reschedule 
      for sub in subs: 
       task = taskqueue.Task(url='/admin/tasks/email', params={'email': sub.emailaddress, 'day': sub.day_no}) 
       task.add("email") 
      cursor = subscribers.cursor() 
      memcache.set('daily_email_cursor', cursor) 
      task = taskqueue.Task(url="/admin/tasks/queue", params={}) 
      task.add("queueup") 

Answer 1

Я вижу пару потенциальных проблем. Во-первых, вы сохраняете свой курсор в memcache, который не гарантирует ничего. Если вы получите пропущенный кэш на полпути через обработку, вы снова отправите все сообщения.

Во-вторых, задачи будут повторно проверены, если они не по какой-либо причине; по этой причине они должны быть предназначены для идемпотента. Конечно, при отправке писем это почти невозможно, так как после отправки сообщения его нельзя отменить, если ваша задача умирает по другой причине после ее отправки. Как минимум, я рекомендую попробовать обновить поле «последняя отправленная дата» на каждом объекте подписчика после отправки им сообщения. Разумеется, это само по себе не является надежным, поскольку отправка электронной почты может быть успешной, и после этого обновление объекта может потерпеть неудачу. Это также добавит накладные расходы на весь процесс, поскольку вы будете писать для каждого подписчика.