Блог

У меня возникли некоторые проблемы с пониманием последовательности событий, вызывающих ошибку в моем приложении, которую можно увидеть только периодически в приложении, развернутом на GAE, и никогда при запуске с локальным devserver.py .

Все связанные фрагменты кода ниже (обрезанные для MCV, надеюсь, я не потерял ничего существенного) выполняются во время обработки одного и того же запроса очереди задач.

Точка входа:

 def job_completed_task(self, _):

    # running outside transaction as query is made
    if not self.all_context_jobs_completed(self.context.db_key, self):
        # this will transactionally enqueue another task
        self.trigger_job_mark_completed_transaction()
    else:
        # this is transactional
        self.context.jobs_completed(self)
  

Соответствующий self.context.jobs_completed(self) :

 @ndb.transactional(xg=True)
def jobs_completed(self, job):

    if self.status == QAStrings.status_done:
        logging.debug('%s jobs_completed %s NOP' % (self.lid, job.job_id))
        return

    # some logic computing step_completed here

    if step_completed:
        self.status = QAStrings.status_done  # includes self.db_data.put()

    # this will transactionally enqueue another task
    job.trigger_job_mark_completed_transaction()
  

self.status Установщик, взломанный для получения обратной трассировки для отладки этого сценария:

 @status.setter
def status(self, new_status):
    assert ndb.in_transaction()

    status = getattr(self, self.attr_status)
    if status != new_status:
        traceback.print_stack()
        logging.info('%s status change %s -> %s' % (self.name, status, new_status))
        setattr(self, self.attr_status, new_status)
  

В job.trigger_job_mark_completed_transaction() конечном итоге ставится в очередь новая задача, подобная этой:

     task = taskqueue.add(queue_name=self.task_queue_name, url=url, params=params,
                         transactional=ndb.in_transaction(), countdown=delay)
  

Журнал GAE для события разделен, поскольку он не помещается на один экран:

Я ожидаю jobs_completed , что транзакция либо увидит ... jobs_completed ... NOP сообщение отладки, а задача не поставлена в очередь, либо, по крайней мере, увидит status change running -> done информационное сообщение и задачу, поставленную в очередь job.trigger_job_mark_completed_transaction() .

На самом деле я вижу как сообщения, так и задачи, не поставленные в очередь.

В журналах отображается, что транзакция выполняется дважды:

• в первый раз он находит статус not done , поэтому выполняет логику, устанавливает статус done равным (и отображает трассировку и сообщение info) и должен транзакционно поставить в очередь новую задачу — но это не так

• 2-й раз он находит статус done и просто печатает сообщение отладки

Мой вопрос — если 1-я попытка транзакции завершается неудачей, не следует ли также откатить изменение статуса? Чего мне не хватает?

Я нашел обходной путь: не указывать повторных попыток jobs_completed() транзакции:

 @ndb.transactional(xg=True, retries=0)
def jobs_completed(self, job):
  

Это предотвращает автоматическое повторное выполнение, вместо этого вызывая исключение:

Ошибка транзакции (транзакция не может быть зафиксирована. Пожалуйста, попробуйте еще раз.)

Что приемлемо, поскольку у меня уже есть система защиты от сбоев / повторных попыток для всего job_completed_task() . Теперь все в порядке.

Что касается того, почему отката не произошло, единственное, что приходит мне в голову, это то, что каким-то образом объект был прочитан (и кэширован в моем атрибуте объекта) до ввода транзакции, поэтому он не считается частью (той же) транзакции. Но я не смог найти путь к коду, который бы это сделал, так что это просто предположение.